forked from Zhubei_iXD/report_skill_expm
Compare commits
No commits in common. "iXD-team" and "main" have entirely different histories.
|
|
@ -41,13 +41,6 @@
|
|||
"english_name": "Shi Ting-Yu",
|
||||
"team": "智能控制組",
|
||||
"department": "軟體部"
|
||||
},
|
||||
{
|
||||
"nickname": "小勳 楊",
|
||||
"formal_name": "楊秉勳",
|
||||
"english_name": "Yang Bing-SYUN",
|
||||
"team": "智能控制組",
|
||||
"department": "軟體部"
|
||||
}
|
||||
],
|
||||
"mapping_config": {
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ Use this skill when:
|
|||
Expected columns in the input file:
|
||||
- `日期` - Date of the work entry
|
||||
- `使用者` - User/person name
|
||||
- `登錄為` - Login as type (Development, Testing, Specification, Support, etc.)
|
||||
- `活動` - Activity type (Development, Testing, Specification, Support, etc.)
|
||||
- `專案` - Project name
|
||||
- `留言` - Comments/notes (optional)
|
||||
- `單位` - Work hours/units
|
||||
|
|
@ -66,7 +66,7 @@ def read_cost_report(file_path):
|
|||
df = pd.read_excel(file_path, header=1)
|
||||
|
||||
# Verify expected columns exist
|
||||
required_cols = ['日期', '使用者', '登錄為', '專案', '單位']
|
||||
required_cols = ['日期', '使用者', '活動', '專案', '單位']
|
||||
if not all(col in df.columns for col in required_cols):
|
||||
# Try header=0 if header=1 doesn't work
|
||||
df = pd.read_excel(file_path, header=0)
|
||||
|
|
@ -77,78 +77,77 @@ def read_cost_report(file_path):
|
|||
return df
|
||||
```
|
||||
|
||||
### Step 2: Process Individual Work Entries by Project
|
||||
### Step 2: Aggregate Data by Person and Project
|
||||
|
||||
```python
|
||||
def aggregate_work_hours(df):
|
||||
"""
|
||||
Process work hours data by project, keeping individual entries separate.
|
||||
Returns a processed DataFrame and date range.
|
||||
Group work hours by user and project.
|
||||
Returns a summary DataFrame.
|
||||
"""
|
||||
# Extract date range for report period
|
||||
df['日期'] = pd.to_datetime(df['日期'])
|
||||
start_date = df['日期'].min()
|
||||
end_date = df['日期'].max()
|
||||
|
||||
# Keep individual work entries separate (no grouping/aggregation)
|
||||
summary = df[['使用者', '專案', '單位', '登錄為', '留言']].copy()
|
||||
summary.columns = ['使用者', '專案', '工時', '登錄為', '備註']
|
||||
# Group by user and project
|
||||
summary = df.groupby(['使用者', '專案', '活動']).agg({
|
||||
'單位': 'sum',
|
||||
'留言': lambda x: '\n'.join(x.dropna().unique()) if '留言' in df.columns else ''
|
||||
}).reset_index()
|
||||
|
||||
# Clean up empty or NaN values
|
||||
summary['登錄為'] = summary['登錄為'].fillna('')
|
||||
summary['備註'] = summary['備註'].fillna('')
|
||||
summary.columns = ['使用者', '專案', '活動', '工時', '備註']
|
||||
|
||||
# Sort by project name and hours (descending)
|
||||
summary = summary.sort_values(['專案', '工時'], ascending=[True, False])
|
||||
# Sort by user and hours (descending)
|
||||
summary = summary.sort_values(['使用者', '工時'], ascending=[True, False])
|
||||
|
||||
return summary, start_date, end_date
|
||||
```
|
||||
|
||||
### Step 3: Format Individual Entries for Weekly Report
|
||||
### Step 3: Format Data for Weekly Report
|
||||
|
||||
```python
|
||||
def format_for_weekly_report(summary_df):
|
||||
"""
|
||||
Transform individual work entries into weekly report format.
|
||||
Group by project but keep individual work entries separate.
|
||||
Transform aggregated data into weekly report format.
|
||||
Group entries by project and list participants.
|
||||
"""
|
||||
# Group by project
|
||||
report_data = []
|
||||
|
||||
for project in summary_df['專案'].unique():
|
||||
project_data = summary_df[summary_df['專案'] == project]
|
||||
|
||||
# Process each work entry individually
|
||||
# Get all participants and their hours
|
||||
participants = []
|
||||
total_hours = 0
|
||||
activities = []
|
||||
notes = []
|
||||
|
||||
for _, row in project_data.iterrows():
|
||||
user_name = row['使用者']
|
||||
user_hours = row['工時']
|
||||
|
||||
# Format progress text from activity and notes
|
||||
progress_parts = []
|
||||
|
||||
if row['登錄為'] and row['登錄為'] != '-':
|
||||
# Extract project name after the colon from format like "Task #838: AXEL CommLib 重構"
|
||||
login_as_text = str(row['登錄為'])
|
||||
if ':' in login_as_text:
|
||||
project_name = login_as_text.split(':', 1)[1].strip()
|
||||
progress_parts.append(f"{project_name}")
|
||||
else:
|
||||
progress_parts.append(f"{login_as_text}")
|
||||
|
||||
participants.append(f"{row['使用者']}")
|
||||
total_hours += row['工時']
|
||||
if row['活動'] and row['活動'] != '-':
|
||||
activities.append(row['活動'])
|
||||
if row['備註']:
|
||||
progress_parts.append(str(row['備註']))
|
||||
|
||||
progress_text = ' '.join(progress_parts) if progress_parts else None
|
||||
|
||||
report_data.append({
|
||||
'專案名稱': project,
|
||||
'子項目名稱': None,
|
||||
'進度': None, # User should fill this in
|
||||
'本周主要進展': progress_text,
|
||||
'參與人員': user_name,
|
||||
'工時': user_hours,
|
||||
'交付物': None,
|
||||
'代碼上傳': None,
|
||||
'下周計畫': None
|
||||
notes.append(row['備註'])
|
||||
|
||||
# Format main progress
|
||||
progress_text = '\n'.join(set(notes)) if notes else ''
|
||||
if activities:
|
||||
activity_text = '\n'.join(set(activities))
|
||||
progress_text = f"{activity_text}\n{progress_text}" if progress_text else activity_text
|
||||
|
||||
report_data.append({
|
||||
'專案名稱': project,
|
||||
'子項目名稱': None,
|
||||
'進度': None, # User should fill this in
|
||||
'本周主要進展': progress_text,
|
||||
'參與人員': ', '.join(set(participants)),
|
||||
'工時': total_hours,
|
||||
'交付物': None,
|
||||
'代碼上傳': None,
|
||||
'下周計畫': None
|
||||
})
|
||||
|
||||
return pd.DataFrame(report_data)
|
||||
|
|
|
|||
|
|
@ -153,39 +153,31 @@ def read_cost_report(file_path):
|
|||
|
||||
def aggregate_work_hours(df):
|
||||
"""
|
||||
Process work hours data by project, keeping individual entries separate.
|
||||
Returns a processed DataFrame and date range.
|
||||
Group work hours by user and project.
|
||||
Returns a summary DataFrame and date range.
|
||||
"""
|
||||
# Extract date range for report period
|
||||
df['日期'] = pd.to_datetime(df['日期'])
|
||||
start_date = df['日期'].min()
|
||||
end_date = df['日期'].max()
|
||||
|
||||
# Prepare the summary DataFrame with individual entries
|
||||
summary_columns = ['使用者', '專案', '工時']
|
||||
# Group by user and project
|
||||
agg_dict = {'單位': 'sum'}
|
||||
|
||||
# Add 登錄為 (login as) column if exists
|
||||
if '登錄為' in df.columns:
|
||||
summary_columns.append('登錄為')
|
||||
else:
|
||||
df['登錄為'] = ''
|
||||
summary_columns.append('登錄為')
|
||||
|
||||
# Add 備註 (comments) column if exists
|
||||
# Add 留言 (comments) if column exists
|
||||
if '留言' in df.columns:
|
||||
summary_columns.append('備註')
|
||||
summary = df[['使用者', '專案', '單位', '登錄為', '留言']].copy()
|
||||
summary.columns = ['使用者', '專案', '工時', '登錄為', '備註']
|
||||
agg_dict['留言'] = lambda x: '\n'.join([str(i) for i in x.dropna().unique() if str(i) != 'nan'])
|
||||
|
||||
# Add 活動 (activity) if column exists
|
||||
if '活動' in df.columns:
|
||||
summary = df.groupby(['使用者', '專案', '活動']).agg(agg_dict).reset_index()
|
||||
summary.columns = ['使用者', '專案', '活動', '工時'] + (['備註'] if '留言' in df.columns else [])
|
||||
else:
|
||||
summary = df[['使用者', '專案', '單位', '登錄為']].copy()
|
||||
summary.columns = ['使用者', '專案', '工時', '登錄為']
|
||||
summary = df.groupby(['使用者', '專案']).agg(agg_dict).reset_index()
|
||||
summary.columns = ['使用者', '專案', '工時'] + (['備註'] if '留言' in df.columns else [])
|
||||
summary['活動'] = ''
|
||||
|
||||
# Clean up empty or NaN values
|
||||
summary['登錄為'] = summary['登錄為'].fillna('')
|
||||
if '備註' in summary.columns:
|
||||
summary['備註'] = summary['備註'].fillna('')
|
||||
|
||||
# Sort by project name first, then by hours (descending) within each project
|
||||
# Sort by user and hours (descending)
|
||||
summary = summary.sort_values(['專案', '工時'], ascending=[True, False])
|
||||
|
||||
return summary, start_date, end_date
|
||||
|
|
@ -193,123 +185,90 @@ def aggregate_work_hours(df):
|
|||
|
||||
def format_for_weekly_report(summary_df, user_manager=None, project_manager=None):
|
||||
"""
|
||||
Transform individual work entries into weekly report format.
|
||||
Group by project and merge entries with same (user, task name) but different comments.
|
||||
Transform aggregated data into weekly report format.
|
||||
Group entries by project and list participants.
|
||||
Uses project hierarchy to organize parent-child project relationships.
|
||||
|
||||
Args:
|
||||
summary_df: Individual work data entries
|
||||
summary_df: Aggregated work data
|
||||
user_manager: Optional UserManager instance for name mapping
|
||||
project_manager: Optional ProjectManager instance for hierarchy
|
||||
"""
|
||||
report_data = []
|
||||
|
||||
# Process projects in order to maintain grouping
|
||||
for project in summary_df['專案'].unique():
|
||||
project_data = summary_df[summary_df['專案'] == project]
|
||||
|
||||
# Group by user and task name for merging
|
||||
grouped_entries = {}
|
||||
# Get all participants and their hours
|
||||
participants = []
|
||||
total_hours = 0
|
||||
activities = []
|
||||
notes = []
|
||||
|
||||
# First pass: collect and group entries
|
||||
for _, row in project_data.iterrows():
|
||||
user_name = row['使用者']
|
||||
user_hours = row['工時']
|
||||
|
||||
# Map nickname to formal name if user manager is available
|
||||
if user_manager:
|
||||
formal_name = user_manager.get_formal_name(user_name, fallback='keep_original')
|
||||
participant = formal_name
|
||||
participants.append(formal_name)
|
||||
else:
|
||||
participant = user_name
|
||||
participants.append(f"{user_name}")
|
||||
|
||||
# Extract task name from 登錄為
|
||||
task_name = None
|
||||
if '登錄為' in row and row['登錄為'] and row['登錄為'] != '-' and pd.notna(row['登錄為']):
|
||||
login_as_text = str(row['登錄為'])
|
||||
if ':' in login_as_text:
|
||||
task_name = login_as_text.split(':', 1)[1].strip()
|
||||
else:
|
||||
task_name = login_as_text
|
||||
total_hours += user_hours
|
||||
|
||||
# Create grouping key
|
||||
group_key = (participant, task_name)
|
||||
if '活動' in row and row['活動'] and row['活動'] != '-' and pd.notna(row['活動']):
|
||||
activities.append(row['活動'])
|
||||
|
||||
# Initialize group if not exists
|
||||
if group_key not in grouped_entries:
|
||||
grouped_entries[group_key] = {
|
||||
'participant': participant,
|
||||
'task_name': task_name,
|
||||
'total_hours': 0,
|
||||
'comments': [],
|
||||
'project': project
|
||||
}
|
||||
|
||||
# Add to group
|
||||
grouped_entries[group_key]['total_hours'] += row['工時']
|
||||
|
||||
# Add comment if exists and not already added
|
||||
if '備註' in row and row['備註'] and pd.notna(row['備註']):
|
||||
comment = str(row['備註']).strip()
|
||||
if comment and comment != "-" and comment not in grouped_entries[group_key]['comments']:
|
||||
grouped_entries[group_key]['comments'].append(comment)
|
||||
notes.append(row['備註'])
|
||||
|
||||
# Second pass: generate report entries from grouped data
|
||||
for group_key, group_data in grouped_entries.items():
|
||||
participant = group_data['participant']
|
||||
task_name = group_data['task_name']
|
||||
total_hours = group_data['total_hours']
|
||||
comments = group_data['comments']
|
||||
# Format main progress
|
||||
progress_text = ''
|
||||
if activities:
|
||||
activity_text = '\n'.join([f"{act}" for act in set(activities)])
|
||||
progress_text = activity_text
|
||||
|
||||
if notes:
|
||||
note_text = '\n'.join(set(notes))
|
||||
progress_text = f"{progress_text}\n{note_text}" if progress_text else note_text
|
||||
|
||||
# Determine if this is a sub-project using project hierarchy
|
||||
parent_project = None
|
||||
project_name = project
|
||||
sub_project_name = None
|
||||
|
||||
if project_manager:
|
||||
# Try to find the project by exact name match
|
||||
project_id = project_manager.find_project_by_name(project)
|
||||
|
||||
# Format progress text
|
||||
progress_parts = []
|
||||
if task_name:
|
||||
progress_parts.append(f"{task_name}")
|
||||
|
||||
# Format comments as bullet points if multiple
|
||||
if comments:
|
||||
if len(comments) == 1:
|
||||
progress_parts.append(comments[0])
|
||||
else:
|
||||
# Multiple comments - format as bullet list
|
||||
bullet_comments = '\n'.join([f" - {comment}" for comment in comments])
|
||||
progress_parts.append(f"\n{bullet_comments}")
|
||||
|
||||
progress_text = ' '.join(progress_parts) if progress_parts else None
|
||||
|
||||
# Determine if this is a sub-project using project hierarchy
|
||||
parent_project = None
|
||||
project_name = project
|
||||
sub_project_name = None
|
||||
|
||||
if project_manager:
|
||||
# Try to find the project by exact name match
|
||||
project_id = project_manager.find_project_by_name(project)
|
||||
if project_id:
|
||||
parent_id = project_manager.get_parent(project_id)
|
||||
|
||||
if project_id:
|
||||
parent_id = project_manager.get_parent(project_id)
|
||||
|
||||
if parent_id:
|
||||
# This is a sub-project
|
||||
parent_info = project_manager.get_project(parent_id)
|
||||
if parent_info:
|
||||
parent_project = parent_info['name']
|
||||
sub_project_name = project
|
||||
project_name = parent_project
|
||||
else:
|
||||
# Project not found in hierarchy, log for debugging
|
||||
pass
|
||||
|
||||
report_data.append({
|
||||
'專案名稱': project_name,
|
||||
'子項目名稱': sub_project_name,
|
||||
'進度': None, # User should fill this in
|
||||
'本周主要進展': progress_text if progress_text else None,
|
||||
'參與人員': participant,
|
||||
'工時': total_hours,
|
||||
'交付物': None,
|
||||
'代碼上傳': None,
|
||||
'下周計畫': None
|
||||
})
|
||||
if parent_id:
|
||||
# This is a sub-project
|
||||
parent_info = project_manager.get_project(parent_id)
|
||||
if parent_info:
|
||||
parent_project = parent_info['name']
|
||||
sub_project_name = project
|
||||
project_name = parent_project
|
||||
else:
|
||||
# Project not found in hierarchy, log for debugging
|
||||
pass
|
||||
|
||||
report_data.append({
|
||||
'專案名稱': project_name,
|
||||
'子項目名稱': sub_project_name,
|
||||
'進度': None, # User should fill this in
|
||||
'本周主要進展': progress_text if progress_text else None,
|
||||
'參與人員': ', '.join(set(participants)),
|
||||
'工時': total_hours,
|
||||
'交付物': None,
|
||||
'代碼上傳': None,
|
||||
'下周計畫': None
|
||||
})
|
||||
|
||||
|
||||
return pd.DataFrame(report_data)
|
||||
|
||||
|
|
@ -359,10 +318,10 @@ def generate_weekly_report(input_file, output_file, template_file=None, team_nam
|
|||
|
||||
summary, start_date, end_date = aggregate_work_hours(df)
|
||||
print(f"Date range: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
||||
print(f"Found {len(summary)} individual work entries across {summary['專案'].nunique()} projects")
|
||||
print(f"Found {len(summary)} work entries across {summary['專案'].nunique()} projects")
|
||||
|
||||
report_df = format_for_weekly_report(summary, user_manager=user_manager, project_manager=project_manager)
|
||||
print(f"Formatted {len(report_df)} individual work entries for report")
|
||||
print(f"Formatted {len(report_df)} project entries for report")
|
||||
|
||||
# Load template or create new workbook
|
||||
if template_file and os.path.exists(template_file):
|
||||
|
|
@ -419,31 +378,8 @@ def generate_weekly_report(input_file, output_file, template_file=None, team_nam
|
|||
|
||||
# Write data starting from row 5
|
||||
current_row = 5
|
||||
project_start_rows = {} # Track where each project starts for merging
|
||||
current_project = None
|
||||
project_start_row = None
|
||||
|
||||
for idx, row_data in report_df.iterrows():
|
||||
project_name = row_data['專案名稱']
|
||||
|
||||
# Check if we're starting a new project
|
||||
if project_name != current_project:
|
||||
# Merge cells for the previous project if it had multiple rows
|
||||
if current_project is not None and project_start_row is not None and current_row > project_start_row + 1:
|
||||
try:
|
||||
ws.merge_cells(f'B{project_start_row}:B{current_row - 1}')
|
||||
# Center the merged cell content
|
||||
merged_cell = ws.cell(row=project_start_row, column=2)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass # Skip if merge fails
|
||||
|
||||
# Start tracking new project
|
||||
current_project = project_name
|
||||
project_start_row = current_row
|
||||
|
||||
# Write data for current row
|
||||
ws.cell(row=current_row, column=2, value=project_name)
|
||||
ws.cell(row=current_row, column=2, value=row_data['專案名稱'])
|
||||
ws.cell(row=current_row, column=3, value=row_data['子項目名稱'])
|
||||
ws.cell(row=current_row, column=4, value=row_data['進度'])
|
||||
ws.cell(row=current_row, column=5, value=row_data['本周主要進展'])
|
||||
|
|
@ -460,38 +396,12 @@ def generate_weekly_report(input_file, output_file, template_file=None, team_nam
|
|||
|
||||
# Add separator row (except after last entry)
|
||||
if idx < len(report_df) - 1:
|
||||
# Check if next entry is different project
|
||||
next_project = report_df.iloc[idx + 1]['專案名稱'] if idx + 1 < len(report_df) else None
|
||||
if next_project != project_name:
|
||||
# Merge cells for current project before adding separator
|
||||
if project_start_row is not None and current_row > project_start_row:
|
||||
try:
|
||||
ws.merge_cells(f'B{project_start_row}:B{current_row - 1}')
|
||||
merged_cell = ws.cell(row=project_start_row, column=2)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Add separator row
|
||||
ws.cell(row=current_row, column=2, value='專案名稱')
|
||||
for col_idx, header in enumerate(headers[1:], start=3):
|
||||
ws.cell(row=current_row, column=col_idx, value=header)
|
||||
# Apply blue header styling to separator row (no blank white spacer)
|
||||
style_header_row(ws, row_idx=current_row, template_cell=template_header_cell)
|
||||
current_row += 1
|
||||
|
||||
# Reset project tracking for next group
|
||||
current_project = None
|
||||
project_start_row = None
|
||||
|
||||
# Handle the last project merging (if needed)
|
||||
if current_project is not None and project_start_row is not None and current_row > project_start_row:
|
||||
try:
|
||||
ws.merge_cells(f'B{project_start_row}:B{current_row - 1}')
|
||||
merged_cell = ws.cell(row=project_start_row, column=2)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass
|
||||
ws.cell(row=current_row, column=2, value='專案名稱')
|
||||
for col_idx, header in enumerate(headers[1:], start=3):
|
||||
ws.cell(row=current_row, column=col_idx, value=header)
|
||||
# Apply blue header styling to separator row (no blank white spacer)
|
||||
style_header_row(ws, row_idx=current_row, template_cell=template_header_cell)
|
||||
current_row += 1
|
||||
|
||||
# Apply formatting (only if no template was used)
|
||||
if not template_sheet:
|
||||
|
|
|
|||
|
|
@ -1,102 +0,0 @@
|
|||
# Week Report Generator from JSON
|
||||
|
||||
這個技能基於 `week_work_package_log` 的 JSON 輸出生成項目週報,與原始的 `week_report_gen` 功能相同但輸入來源不同。
|
||||
|
||||
## 主要差異
|
||||
|
||||
### 輸入格式
|
||||
- **原始 week_report_gen**: Excel 成本報告檔案 (cost-report-*.xls)
|
||||
- **新 week_report_gen_from_json**: JSON 格式的工時數據
|
||||
|
||||
### 數據來源
|
||||
- **原始**: 手動匯出的 Excel 檔案
|
||||
- **新**: 直接從 OpenProject API 獲取的結構化數據
|
||||
|
||||
### 欄位邏輯
|
||||
按照新需求實現的邏輯:
|
||||
|
||||
#### B欄 (專案名稱)
|
||||
顯示最上級專案名稱(`type: "project"` 的節點)
|
||||
|
||||
#### C欄 (子項目名稱)
|
||||
- 如果底層任務屬於子專案 → 顯示最上級的 `work_package`
|
||||
- 如果沒有中間層級 → 顯示 `"-"`
|
||||
|
||||
#### D欄 (進度)
|
||||
跟隨 C 欄合併,因為進度是針對整個子項目的狀態
|
||||
|
||||
#### E欄 (本周主要進展)
|
||||
顯示最底層工作包名稱(有 `time_entries` 的節點),如果有 comment 也一併顯示
|
||||
|
||||
## 使用方法
|
||||
|
||||
### 1. 指定日期範圍(自動獲取數據)
|
||||
```bash
|
||||
python generate_report.py "2026-01-20" "2026-01-22" "output.xlsx"
|
||||
```
|
||||
|
||||
### 2. 使用已有的 JSON 檔案
|
||||
```bash
|
||||
python generate_report.py "data.json" "output.xlsx"
|
||||
```
|
||||
|
||||
### 3. 使用預設值(本週數據)
|
||||
```bash
|
||||
python generate_report.py
|
||||
```
|
||||
|
||||
## 功能特色
|
||||
|
||||
- ✓ 自動從 OpenProject API 獲取數據
|
||||
- ✓ 支援使用者名稱映射 (UserManager)
|
||||
- ✓ 階層專案結構識別
|
||||
- ✓ 工時聚合和評論合併
|
||||
- ✓ 智能儲存格合併邏輯
|
||||
- **B 欄**: 相同專案名稱合併
|
||||
- **C 欄**: 相同子項目名稱合併("-" 不合併)
|
||||
- **D 欄**: 跟隨 C 欄合併(進度針對子項目)
|
||||
- ✓ 使用相同模板格式
|
||||
- ✓ 完整的樣式和格式化
|
||||
|
||||
## 排序邏輯
|
||||
|
||||
數據按以下順序排序:
|
||||
1. **專案名稱** (升序)
|
||||
2. **子項目名稱** (升序)
|
||||
3. **工時** (降序)
|
||||
|
||||
這確保相同專案和子項目的條目會聚集在一起,便於合併儲存格。
|
||||
|
||||
## 合併邏輯
|
||||
|
||||
### 專案層級合併 (B 欄)
|
||||
- 相同專案名稱的多行會合併 B 欄儲存格
|
||||
- 合併儲存格垂直置中對齊
|
||||
|
||||
### 子項目層級合併 (C & D 欄)
|
||||
- 相同子項目名稱的多行會同時合併 C 欄和 D 欄儲存格
|
||||
- **特殊處理**: 子項目名稱為 `"-"` 的條目不會合併(因為 "-" 表示沒有子項目,每個都是獨立的)
|
||||
- 合併儲存格垂直置中對齊
|
||||
|
||||
## 依賴技能
|
||||
|
||||
- `week_work_package_log`: 獲取 JSON 數據
|
||||
- `list_user`: 使用者名稱映射 (可選)
|
||||
- `week_report_gen`: 模板檔案共用
|
||||
|
||||
## 輸出格式
|
||||
|
||||
生成的 Excel 檔案格式與原始 `week_report_gen` 完全相同:
|
||||
- 公司標題和日期範圍
|
||||
- 標準化的欄位結構
|
||||
- 專案分組和合併儲存格
|
||||
- 一致的顏色和邊框樣式
|
||||
- 專業的視覺呈現效果
|
||||
|
||||
## 範例輸出
|
||||
|
||||
生成的週報會包含:
|
||||
- **專案分組**: 每個專案的條目會聚集在一起
|
||||
- **子項目合併**: 相同子項目的進度欄位統一顯示
|
||||
- **清晰的層次**: 從專案 → 子項目 → 具體任務的清晰結構
|
||||
- **工時統計**: 每個任務的詳細工時記錄
|
||||
|
|
@ -1,164 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Week Report Generator from JSON 使用範例腳本
|
||||
展示基於 JSON 數據生成週報的各種使用場景
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Add the skill to Python path
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, script_dir)
|
||||
|
||||
from generate_report import generate_weekly_report_from_json
|
||||
|
||||
|
||||
def example_this_week():
|
||||
"""範例1: 生成本週週報(自動獲取數據)"""
|
||||
|
||||
today = datetime.now()
|
||||
days_since_monday = today.weekday()
|
||||
this_monday = today - timedelta(days=days_since_monday)
|
||||
this_friday = this_monday + timedelta(days=4)
|
||||
|
||||
start_date = this_monday.strftime("%Y-%m-%d")
|
||||
end_date = this_friday.strftime("%Y-%m-%d")
|
||||
|
||||
print(f"📅 範例1: 本週週報生成 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
output_file = f"temp/週報-本週-{datetime.now().strftime('%Y%m%d')}.xlsx"
|
||||
|
||||
output_path, report_df = generate_weekly_report_from_json(
|
||||
json_input=(start_date, end_date),
|
||||
output_file=output_file
|
||||
)
|
||||
|
||||
print(f"\n✅ 週報生成成功!")
|
||||
print(f"📁 輸出檔案: {output_path}")
|
||||
print(f"📊 包含 {len(report_df)} 個工時記錄")
|
||||
print(f"🏢 涵蓋 {report_df['專案名稱'].nunique()} 個專案")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_specific_dates():
|
||||
"""範例2: 生成指定日期範圍的週報"""
|
||||
|
||||
start_date = "2026-01-20"
|
||||
end_date = "2026-01-22"
|
||||
|
||||
print(f"📅 範例2: 指定日期週報 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
output_file = f"temp/週報-指定日期-{end_date.replace('-', '')}.xlsx"
|
||||
|
||||
output_path, report_df = generate_weekly_report_from_json(
|
||||
json_input=(start_date, end_date),
|
||||
output_file=output_file,
|
||||
team_name="智能控制組"
|
||||
)
|
||||
|
||||
print(f"\n✅ 週報生成成功!")
|
||||
print(f"📁 輸出檔案: {output_path}")
|
||||
|
||||
# 顯示專案摘要
|
||||
print(f"\n📋 專案摘要:")
|
||||
for project in report_df['專案名稱'].unique():
|
||||
project_data = report_df[report_df['專案名稱'] == project]
|
||||
total_hours = project_data['工時'].sum()
|
||||
print(f" • {project}: {total_hours} 小時 ({len(project_data)} 個任務)")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_with_json_file():
|
||||
"""範例3: 從現有 JSON 檔案生成週報"""
|
||||
|
||||
print("📅 範例3: 從 JSON 檔案生成週報")
|
||||
print("=" * 70)
|
||||
|
||||
# 先生成 JSON 數據(模擬已有的數據文件)
|
||||
from week_work_package_log.work_package_logger import WeekWorkPackageLogger
|
||||
|
||||
try:
|
||||
# 獲取 JSON 數據
|
||||
logger = WeekWorkPackageLogger()
|
||||
json_data = logger.fetch_and_display_time_entries("2026-01-20", "2026-01-22", 'json')
|
||||
|
||||
if json_data:
|
||||
# 保存為檔案
|
||||
import json
|
||||
json_file = "temp/work_data.json"
|
||||
with open(json_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(json_data, f, ensure_ascii=False, indent=2)
|
||||
|
||||
print(f"💾 JSON 數據已保存到: {json_file}")
|
||||
|
||||
# 從 JSON 檔案生成週報
|
||||
output_file = "temp/週報-from-json-file.xlsx"
|
||||
output_path, report_df = generate_weekly_report_from_json(
|
||||
json_input=json_file,
|
||||
output_file=output_file
|
||||
)
|
||||
|
||||
print(f"\n✅ 從 JSON 檔案生成週報成功!")
|
||||
print(f"📁 輸出檔案: {output_path}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def show_features():
|
||||
"""展示主要功能特色"""
|
||||
|
||||
print("🚀 Week Report Generator from JSON 功能特色")
|
||||
print("=" * 70)
|
||||
print("✓ 直接從 OpenProject API 獲取數據")
|
||||
print("✓ 智能儲存格合併:")
|
||||
print(" - B 欄:相同專案名稱自動合併")
|
||||
print(" - C 欄:相同子項目名稱自動合併")
|
||||
print(" - D 欄:跟隨子項目進度合併")
|
||||
print("✓ 階層專案結構識別")
|
||||
print("✓ 使用者名稱自動映射")
|
||||
print("✓ 專業週報模板格式")
|
||||
print("✓ 支援多種輸入方式:日期範圍、JSON檔案、預設值")
|
||||
print("✓ 自動排序和分組")
|
||||
print("✓ 工時聚合和評論合併")
|
||||
|
||||
|
||||
def main():
|
||||
"""主函數 - 展示所有範例"""
|
||||
|
||||
print("🎯 Week Report Generator from JSON 使用範例")
|
||||
print("=" * 70)
|
||||
|
||||
# 確保 temp 目錄存在
|
||||
os.makedirs("temp", exist_ok=True)
|
||||
|
||||
# 展示功能特色
|
||||
show_features()
|
||||
print("\n")
|
||||
|
||||
# 執行範例
|
||||
example_this_week()
|
||||
print("\n")
|
||||
|
||||
example_specific_dates()
|
||||
print("\n")
|
||||
|
||||
example_with_json_file()
|
||||
print("\n")
|
||||
|
||||
print("🎉 所有範例執行完畢!")
|
||||
print("📁 請檢查 temp/ 目錄中生成的週報檔案")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,770 +0,0 @@
|
|||
"""
|
||||
Weekly Report Generator from JSON
|
||||
|
||||
This script generates weekly project reports from week_work_package_log JSON output.
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
import openpyxl
|
||||
import json
|
||||
from datetime import datetime
|
||||
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
|
||||
from copy import copy
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Add list_user skill to path for user mapping
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
skills_dir = os.path.dirname(script_dir) # .../skills
|
||||
list_user_path = os.path.join(skills_dir, 'list_user')
|
||||
if list_user_path not in sys.path:
|
||||
sys.path.insert(0, list_user_path)
|
||||
|
||||
try:
|
||||
from user_manager import UserManager
|
||||
USER_MANAGER_AVAILABLE = True
|
||||
except ImportError as e:
|
||||
USER_MANAGER_AVAILABLE = False
|
||||
|
||||
# Add week_work_package_log skill to path for JSON data generation
|
||||
work_package_log_path = os.path.join(skills_dir, 'week_work_package_log')
|
||||
if work_package_log_path not in sys.path:
|
||||
sys.path.insert(0, work_package_log_path)
|
||||
|
||||
try:
|
||||
from work_package_logger import WeekWorkPackageLogger
|
||||
WORK_PACKAGE_LOG_AVAILABLE = True
|
||||
except ImportError as e:
|
||||
WORK_PACKAGE_LOG_AVAILABLE = False
|
||||
|
||||
|
||||
# Style constants for consistent coloring
|
||||
HEADER_FILL_COLOR = 'FF2F75B5' # Blue header background
|
||||
DATA_FILL_COLOR = 'FFE7E6E6' # Light gray data background
|
||||
WHITE_BORDER = Border(
|
||||
left=Side(style='thin', color='FFFFFFFF'),
|
||||
right=Side(style='thin', color='FFFFFFFF'),
|
||||
top=Side(style='thin', color='FFFFFFFF'),
|
||||
bottom=Side(style='thin', color='FFFFFFFF')
|
||||
)
|
||||
|
||||
|
||||
def copy_cell_style(source_cell, target_cell):
|
||||
"""
|
||||
Copy all styling from source cell to target cell.
|
||||
"""
|
||||
if source_cell.has_style:
|
||||
target_cell.font = copy(source_cell.font)
|
||||
target_cell.border = copy(source_cell.border)
|
||||
target_cell.fill = copy(source_cell.fill)
|
||||
target_cell.number_format = copy(source_cell.number_format)
|
||||
target_cell.protection = copy(source_cell.protection)
|
||||
target_cell.alignment = copy(source_cell.alignment)
|
||||
|
||||
|
||||
def copy_sheet_with_format(source_sheet, target_sheet, max_rows=10):
|
||||
"""
|
||||
Copy formatting from source sheet template to target sheet.
|
||||
Only copies the first few rows (headers) to preserve template styling.
|
||||
"""
|
||||
# Copy row heights
|
||||
for row_idx in range(1, max_rows + 1):
|
||||
if row_idx in source_sheet.row_dimensions:
|
||||
target_sheet.row_dimensions[row_idx].height = source_sheet.row_dimensions[row_idx].height
|
||||
|
||||
# Copy column widths
|
||||
for col_letter in ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K']:
|
||||
if col_letter in source_sheet.column_dimensions:
|
||||
target_sheet.column_dimensions[col_letter].width = source_sheet.column_dimensions[col_letter].width
|
||||
|
||||
# Copy cell formatting for header rows (first max_rows)
|
||||
for row_idx in range(1, max_rows + 1):
|
||||
for col_idx in range(1, 12): # Columns A-K
|
||||
source_cell = source_sheet.cell(row=row_idx, column=col_idx)
|
||||
target_cell = target_sheet.cell(row=row_idx, column=col_idx)
|
||||
|
||||
# Copy style
|
||||
copy_cell_style(source_cell, target_cell)
|
||||
|
||||
# Copy merged cells info (we'll handle merging separately)
|
||||
|
||||
# Copy merged cells from template (only header rows)
|
||||
for merged_range in source_sheet.merged_cells.ranges:
|
||||
if merged_range.min_row <= max_rows and merged_range.max_row <= max_rows:
|
||||
try:
|
||||
target_sheet.merge_cells(str(merged_range))
|
||||
except:
|
||||
pass # Skip if already merged or invalid
|
||||
|
||||
|
||||
def style_header_row(ws, row_idx, template_cell=None, col_start=2, col_end=10):
|
||||
"""
|
||||
Apply header styling (blue fill, bold, centered, white grid) to a row.
|
||||
If template_cell is provided, its style is copied before applying borders.
|
||||
"""
|
||||
for col_idx in range(col_start, col_end + 1):
|
||||
cell = ws.cell(row=row_idx, column=col_idx)
|
||||
if template_cell and template_cell.has_style:
|
||||
copy_cell_style(template_cell, cell)
|
||||
else:
|
||||
cell.fill = PatternFill(start_color=HEADER_FILL_COLOR, end_color=HEADER_FILL_COLOR, fill_type='solid')
|
||||
cell.font = Font(bold=True)
|
||||
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
cell.border = WHITE_BORDER
|
||||
|
||||
|
||||
def style_data_row(ws, row_idx, fill_color=DATA_FILL_COLOR, col_start=2, col_end=10):
|
||||
"""
|
||||
Apply data row styling (fill, wrap text, white grid) to a row.
|
||||
"""
|
||||
data_fill = PatternFill(start_color=fill_color, end_color=fill_color, fill_type='solid')
|
||||
data_align = Alignment(vertical='top', wrap_text=True)
|
||||
for col_idx in range(col_start, col_end + 1):
|
||||
cell = ws.cell(row=row_idx, column=col_idx)
|
||||
cell.fill = data_fill
|
||||
cell.alignment = data_align
|
||||
cell.border = WHITE_BORDER
|
||||
|
||||
|
||||
def read_json_data(json_input):
|
||||
"""
|
||||
Read JSON data from file or string and parse the structure.
|
||||
|
||||
Args:
|
||||
json_input: Either file path to JSON file or JSON string
|
||||
|
||||
Returns:
|
||||
dict: Parsed JSON data
|
||||
"""
|
||||
try:
|
||||
if os.path.exists(json_input):
|
||||
# Input is a file path
|
||||
with open(json_input, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
else:
|
||||
# Input is a JSON string
|
||||
data = json.loads(json_input)
|
||||
|
||||
# Validate JSON structure
|
||||
if 'projects' not in data or 'summary' not in data:
|
||||
raise ValueError("Invalid JSON format: missing 'projects' or 'summary' key")
|
||||
|
||||
return data
|
||||
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"Error reading JSON data: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"Error processing JSON data: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def find_leaf_work_packages(node, path=[]):
|
||||
"""
|
||||
Recursively find all leaf work packages (those with time_entries).
|
||||
|
||||
Args:
|
||||
node: Current node in the tree
|
||||
path: Current path from root to this node
|
||||
|
||||
Returns:
|
||||
list: List of (leaf_node, path_to_leaf) tuples
|
||||
"""
|
||||
current_path = path + [node]
|
||||
leaf_packages = []
|
||||
|
||||
# If this node has time_entries, it's a leaf
|
||||
if node.get('time_entries', []):
|
||||
leaf_packages.append((node, current_path))
|
||||
|
||||
# Recursively search children
|
||||
for child in node.get('children', []):
|
||||
leaf_packages.extend(find_leaf_work_packages(child, current_path))
|
||||
|
||||
return leaf_packages
|
||||
|
||||
|
||||
def extract_report_data_from_json(json_data, user_manager=None):
|
||||
"""
|
||||
Extract report data from JSON structure following the new logic.
|
||||
|
||||
Args:
|
||||
json_data: Parsed JSON data from week_work_package_log
|
||||
user_manager: Optional UserManager for name mapping
|
||||
|
||||
Returns:
|
||||
tuple: (report_data_list, start_date, end_date)
|
||||
"""
|
||||
report_data = []
|
||||
|
||||
# Extract date range from summary
|
||||
date_range = json_data['summary']['date_range']
|
||||
start_date = datetime.strptime(date_range.split(' to ')[0], '%Y-%m-%d')
|
||||
end_date = datetime.strptime(date_range.split(' to ')[1], '%Y-%m-%d')
|
||||
|
||||
# Process each project
|
||||
for project in json_data['projects']:
|
||||
# Find all leaf work packages in this project
|
||||
leaf_packages = find_leaf_work_packages(project)
|
||||
|
||||
for leaf_node, path_to_leaf in leaf_packages:
|
||||
# Process each time entry in the leaf node
|
||||
for time_entry in leaf_node['time_entries']:
|
||||
user_name = time_entry['user']
|
||||
|
||||
# Map nickname to formal name if user manager is available
|
||||
if user_manager:
|
||||
formal_name = user_manager.get_formal_name(user_name, fallback='keep_original')
|
||||
participant = formal_name
|
||||
else:
|
||||
participant = user_name
|
||||
|
||||
# Determine project name (B column) - always the root project
|
||||
project_name = path_to_leaf[0]['name'] # First element is always the project
|
||||
|
||||
# Determine sub-project name (C column)
|
||||
# Structure: project_head -> project_1-1 -> ... -> project_1-1-3-2 -> work_package_1 -> work_package_1-1 -> ... -> work_package_end -> leaf
|
||||
# Priority: 1) First work_package (highest level), 2) Last project (lowest level), 3) "-"
|
||||
sub_project_name = "-"
|
||||
first_work_package = None
|
||||
first_work_package_index = None
|
||||
last_project = None
|
||||
last_project_index = None
|
||||
|
||||
# Skip the first node (root project) and last node (leaf), track level indices
|
||||
for level_index, node in enumerate(path_to_leaf[1:-1], start=1):
|
||||
if node['type'] == 'work_package' and first_work_package is None:
|
||||
first_work_package = node['name'] # Take the first work_package found
|
||||
first_work_package_index = level_index
|
||||
elif node['type'] == 'project':
|
||||
last_project = node['name'] # Keep updating to get the last project
|
||||
last_project_index = level_index
|
||||
|
||||
# Apply priority logic and extract completion info
|
||||
completion = None
|
||||
work_package_node = None
|
||||
|
||||
if first_work_package:
|
||||
sub_project_name = first_work_package # Highest priority: first work_package
|
||||
# Find the work package node to extract completion
|
||||
for node in path_to_leaf[1:-1]:
|
||||
if node['type'] == 'work_package' and node['name'] == first_work_package:
|
||||
work_package_node = node
|
||||
break
|
||||
elif last_project:
|
||||
sub_project_name = last_project # Second priority: last project
|
||||
|
||||
# Extract completion from work package if available
|
||||
if work_package_node and 'completion' in work_package_node:
|
||||
completion = work_package_node['completion']
|
||||
|
||||
# If C column is empty or "-", try to get completion from leaf work package (E column)
|
||||
if (sub_project_name == "-" or not sub_project_name) and leaf_node.get('completion'):
|
||||
completion = leaf_node['completion']
|
||||
|
||||
# Task name (E column) - leaf work package name
|
||||
task_name = leaf_node['name']
|
||||
|
||||
# Format progress text with comments
|
||||
progress_parts = [task_name]
|
||||
if time_entry.get('comment') and time_entry['comment'].strip():
|
||||
progress_parts.append(f" - {time_entry['comment'].strip()}")
|
||||
|
||||
progress_text = '\n'.join(progress_parts)
|
||||
|
||||
report_data.append({
|
||||
'專案名稱': project_name,
|
||||
'子項目名稱': sub_project_name,
|
||||
'進度': completion, # Use work package completion if available
|
||||
'本周主要進展': progress_text,
|
||||
'參與人員': participant,
|
||||
'工時': time_entry['hours'],
|
||||
'交付物': None,
|
||||
'代碼上傳': None,
|
||||
'下周計畫': None
|
||||
})
|
||||
|
||||
return report_data, start_date, end_date
|
||||
|
||||
|
||||
def format_for_weekly_report(json_data, user_manager=None):
|
||||
"""
|
||||
Transform JSON data into weekly report format with aggregation.
|
||||
Groups by project, sub-project, user, and task, merging comments.
|
||||
|
||||
Args:
|
||||
json_data: Parsed JSON data from week_work_package_log
|
||||
user_manager: Optional UserManager instance for name mapping
|
||||
|
||||
Returns:
|
||||
pd.DataFrame: Formatted report data
|
||||
"""
|
||||
# Extract individual entries
|
||||
report_entries, start_date, end_date = extract_report_data_from_json(json_data, user_manager)
|
||||
|
||||
# Group entries for aggregation
|
||||
grouped_entries = {}
|
||||
|
||||
for entry in report_entries:
|
||||
# Create grouping key: project + sub-project + user + task
|
||||
group_key = (
|
||||
entry['專案名稱'],
|
||||
entry['子項目名稱'],
|
||||
entry['參與人員'],
|
||||
entry['本周主要進展'].split('\n')[0] # Task name (first line)
|
||||
)
|
||||
|
||||
if group_key not in grouped_entries:
|
||||
grouped_entries[group_key] = {
|
||||
'專案名稱': entry['專案名稱'],
|
||||
'子項目名稱': entry['子項目名稱'],
|
||||
'進度': entry['進度'], # Use completion from the first entry in the group
|
||||
'本周主要進展': [entry['本周主要進展'].split('\n')[0]], # Task name
|
||||
'參與人員': entry['參與人員'],
|
||||
'工時': 0,
|
||||
'交付物': None,
|
||||
'代碼上傳': None,
|
||||
'下周計畫': None,
|
||||
'comments': []
|
||||
}
|
||||
|
||||
# Update completion if the current entry has completion info and grouped entry doesn't
|
||||
if entry['進度'] is not None and grouped_entries[group_key]['進度'] is None:
|
||||
grouped_entries[group_key]['進度'] = entry['進度']
|
||||
|
||||
# Aggregate hours
|
||||
grouped_entries[group_key]['工時'] += entry['工時']
|
||||
|
||||
# Collect comments (skip task name)
|
||||
entry_lines = entry['本周主要進展'].split('\n')
|
||||
if len(entry_lines) > 1: # Has comments beyond task name
|
||||
for comment in entry_lines[1:]:
|
||||
comment = comment.strip()
|
||||
if comment and comment not in grouped_entries[group_key]['comments']:
|
||||
grouped_entries[group_key]['comments'].append(comment)
|
||||
|
||||
# Format final entries
|
||||
final_report_data = []
|
||||
for group_key, group_data in grouped_entries.items():
|
||||
# Format progress text: task name + comments
|
||||
progress_parts = group_data['本周主要進展'] # Task name
|
||||
if group_data['comments']:
|
||||
progress_parts.extend(group_data['comments'])
|
||||
|
||||
progress_text = '\n'.join(progress_parts)
|
||||
|
||||
final_report_data.append({
|
||||
'專案名稱': group_data['專案名稱'],
|
||||
'子項目名稱': group_data['子項目名稱'],
|
||||
'進度': group_data['進度'],
|
||||
'本周主要進展': progress_text,
|
||||
'參與人員': group_data['參與人員'],
|
||||
'工時': group_data['工時'],
|
||||
'交付物': group_data['交付物'],
|
||||
'代碼上傳': group_data['代碼上傳'],
|
||||
'下周計畫': group_data['下周計畫']
|
||||
})
|
||||
|
||||
# Sort by project name first, then by sub-project name, then by hours (descending)
|
||||
final_report_data.sort(key=lambda x: (x['專案名稱'], x['子項目名稱'], -x['工時']))
|
||||
|
||||
return pd.DataFrame(final_report_data), start_date, end_date
|
||||
|
||||
|
||||
def fetch_json_data(start_date, end_date):
|
||||
"""
|
||||
Fetch JSON data using week_work_package_log skill.
|
||||
|
||||
Args:
|
||||
start_date: Start date in YYYY-MM-DD format
|
||||
end_date: End date in YYYY-MM-DD format
|
||||
|
||||
Returns:
|
||||
dict: JSON data from work package logger
|
||||
"""
|
||||
if not WORK_PACKAGE_LOG_AVAILABLE:
|
||||
raise ImportError("week_work_package_log skill not available")
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
json_data = logger.fetch_and_display_time_entries(start_date, end_date, 'json')
|
||||
|
||||
if not json_data:
|
||||
raise ValueError(f"No data returned for date range {start_date} to {end_date}")
|
||||
|
||||
return json_data
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error fetching work package data: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def generate_weekly_report_from_json(json_input, output_file, template_file=None, team_name="智能控制組", use_user_mapping=True):
|
||||
"""
|
||||
Main function to generate weekly report from JSON data.
|
||||
|
||||
Args:
|
||||
json_input: JSON file path, JSON string, or tuple (start_date, end_date) to fetch data
|
||||
output_file: Path to output weekly report Excel file
|
||||
template_file: Optional path to template file
|
||||
team_name: Team name for sheet title (default: "智能控制組")
|
||||
use_user_mapping: Whether to apply user name mapping (default: True)
|
||||
|
||||
Returns:
|
||||
Tuple of (output_file_path, report_dataframe)
|
||||
"""
|
||||
# Initialize user manager for name mapping
|
||||
user_manager = None
|
||||
if use_user_mapping and USER_MANAGER_AVAILABLE:
|
||||
try:
|
||||
user_manager = UserManager()
|
||||
print("✓ User name mapping enabled")
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not load user mapping: {e}")
|
||||
print(" Continuing without user name mapping...")
|
||||
elif use_user_mapping and not USER_MANAGER_AVAILABLE:
|
||||
print("⚠ Warning: list_user skill not available, skipping user name mapping")
|
||||
|
||||
# Get JSON data
|
||||
if isinstance(json_input, tuple) and len(json_input) == 2:
|
||||
# Fetch data using dates
|
||||
start_date, end_date = json_input
|
||||
print(f"Fetching work package data from {start_date} to {end_date}...")
|
||||
json_data = fetch_json_data(start_date, end_date)
|
||||
else:
|
||||
# Read from file or string
|
||||
print(f"Reading JSON data from: {json_input}")
|
||||
json_data = read_json_data(json_input)
|
||||
|
||||
print(f"Loaded JSON data with {len(json_data['projects'])} projects")
|
||||
|
||||
# Process data
|
||||
report_df, start_date, end_date = format_for_weekly_report(json_data, user_manager=user_manager)
|
||||
print(f"Date range: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
||||
print(f"Formatted {len(report_df)} work entries for report")
|
||||
|
||||
# Load template or create new workbook
|
||||
if template_file and os.path.exists(template_file):
|
||||
print(f"Loading template from: {template_file}")
|
||||
wb = openpyxl.load_workbook(template_file)
|
||||
|
||||
# Get the template sheet
|
||||
template_sheet = None
|
||||
for sheet_name_candidate in ['周报示例', wb.sheetnames[0]]:
|
||||
if sheet_name_candidate in wb.sheetnames:
|
||||
template_sheet = wb[sheet_name_candidate]
|
||||
break
|
||||
|
||||
# Create new sheet for this week
|
||||
sheet_name = f"{end_date.strftime('%Y%m%d')}-{team_name}"
|
||||
ws = wb.create_sheet(sheet_name, 0) # Insert at beginning
|
||||
|
||||
# Copy formatting from template if available
|
||||
if template_sheet:
|
||||
print(f"Copying formatting from template sheet: {template_sheet.title}")
|
||||
copy_sheet_with_format(template_sheet, ws, max_rows=4)
|
||||
|
||||
# Unmerge cells beyond row 4 to allow data writing
|
||||
merged_ranges_to_remove = []
|
||||
for merged_range in ws.merged_cells.ranges:
|
||||
if merged_range.min_row > 4:
|
||||
merged_ranges_to_remove.append(merged_range)
|
||||
|
||||
for merged_range in merged_ranges_to_remove:
|
||||
ws.unmerge_cells(str(merged_range))
|
||||
else:
|
||||
print("Creating new workbook (no template provided)")
|
||||
wb = openpyxl.Workbook()
|
||||
ws = wb.active
|
||||
sheet_name = f"{end_date.strftime('%Y%m%d')}-{team_name}"
|
||||
ws.title = sheet_name
|
||||
template_sheet = None
|
||||
|
||||
# Write header
|
||||
ws['B1'] = '弘訊科技股份有限公司'
|
||||
ws['B2'] = f"2026年度週報({start_date.strftime('%m月%d日')}-{end_date.strftime('%m月%d日')})"
|
||||
ws['B3'] = '建議用可視化素材(如有)輔助報告:設計圖、產品架構圖、產品實物照片等等相關材料'
|
||||
|
||||
# Write column headers (row 4)
|
||||
headers = ['專案名稱', '子項目名稱', '進度', '本周主要進展 (請條列說明)',
|
||||
'參與人員', '工时(个人投入研发工时', '交付物', '代碼(如有)是否上傳', '下周計畫']
|
||||
for col_idx, header in enumerate(headers, start=2): # Start from column B
|
||||
ws.cell(row=4, column=col_idx, value=header)
|
||||
|
||||
# Ensure header row is uniformly blue
|
||||
template_header_cell = template_sheet.cell(row=4, column=2) if template_sheet else None
|
||||
style_header_row(ws, row_idx=4, template_cell=template_header_cell)
|
||||
|
||||
# Write data starting from row 5
|
||||
current_row = 5
|
||||
current_project = None
|
||||
current_sub_project = None
|
||||
project_start_row = None
|
||||
sub_project_start_row = None
|
||||
|
||||
for idx, row_data in report_df.iterrows():
|
||||
project_name = row_data['專案名稱']
|
||||
sub_project_name = row_data['子項目名稱']
|
||||
|
||||
# Check if we're starting a new project
|
||||
if project_name != current_project:
|
||||
# Merge cells for the previous project if it had multiple rows
|
||||
if current_project is not None and project_start_row is not None and current_row > project_start_row + 1:
|
||||
try:
|
||||
ws.merge_cells(f'B{project_start_row}:B{current_row - 1}')
|
||||
# Center the merged cell content
|
||||
merged_cell = ws.cell(row=project_start_row, column=2)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass # Skip if merge fails
|
||||
|
||||
# Merge cells for the previous sub-project if it had multiple rows and is not "-"
|
||||
if (current_sub_project is not None and current_sub_project != "-" and
|
||||
sub_project_start_row is not None and current_row > sub_project_start_row + 1):
|
||||
try:
|
||||
# Merge C column (sub-project name)
|
||||
ws.merge_cells(f'C{sub_project_start_row}:C{current_row - 1}')
|
||||
merged_cell = ws.cell(row=sub_project_start_row, column=3)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
|
||||
# Merge D column (progress) as well
|
||||
ws.merge_cells(f'D{sub_project_start_row}:D{current_row - 1}')
|
||||
merged_cell_d = ws.cell(row=sub_project_start_row, column=4)
|
||||
merged_cell_d.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass # Skip if merge fails
|
||||
|
||||
# Start tracking new project
|
||||
current_project = project_name
|
||||
project_start_row = current_row
|
||||
current_sub_project = sub_project_name
|
||||
sub_project_start_row = current_row
|
||||
|
||||
# Check if we're starting a new sub-project (within same project)
|
||||
elif sub_project_name != current_sub_project:
|
||||
# Merge cells for the previous sub-project if it had multiple rows and is not "-"
|
||||
if (current_sub_project is not None and current_sub_project != "-" and
|
||||
sub_project_start_row is not None and current_row > sub_project_start_row + 1):
|
||||
try:
|
||||
# Merge C column (sub-project name)
|
||||
ws.merge_cells(f'C{sub_project_start_row}:C{current_row - 1}')
|
||||
merged_cell = ws.cell(row=sub_project_start_row, column=3)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
|
||||
# Merge D column (progress) as well
|
||||
ws.merge_cells(f'D{sub_project_start_row}:D{current_row - 1}')
|
||||
merged_cell_d = ws.cell(row=sub_project_start_row, column=4)
|
||||
merged_cell_d.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass # Skip if merge fails
|
||||
|
||||
# Start tracking new sub-project
|
||||
current_sub_project = sub_project_name
|
||||
sub_project_start_row = current_row
|
||||
|
||||
# Write data for current row
|
||||
ws.cell(row=current_row, column=2, value=project_name)
|
||||
ws.cell(row=current_row, column=3, value=row_data['子項目名稱'])
|
||||
|
||||
# Handle progress column (D) with percentage formatting
|
||||
progress_cell = ws.cell(row=current_row, column=4, value=row_data['進度'])
|
||||
if row_data['進度'] and str(row_data['進度']).strip():
|
||||
# Try to extract numeric value and format as percentage
|
||||
progress_value = str(row_data['進度']).strip()
|
||||
if progress_value.endswith('%'):
|
||||
try:
|
||||
numeric_value = float(progress_value[:-1]) / 100
|
||||
progress_cell.value = numeric_value
|
||||
progress_cell.number_format = '0%'
|
||||
except ValueError:
|
||||
# If conversion fails, keep original text
|
||||
pass
|
||||
|
||||
ws.cell(row=current_row, column=5, value=row_data['本周主要進展'])
|
||||
ws.cell(row=current_row, column=6, value=row_data['參與人員'])
|
||||
ws.cell(row=current_row, column=7, value=row_data['工時'])
|
||||
ws.cell(row=current_row, column=8, value=row_data['交付物'])
|
||||
ws.cell(row=current_row, column=9, value=row_data['代碼上傳'])
|
||||
ws.cell(row=current_row, column=10, value=row_data['下周計畫'])
|
||||
|
||||
# Apply consistent gray fill and white borders to data row
|
||||
style_data_row(ws, row_idx=current_row)
|
||||
|
||||
current_row += 1
|
||||
|
||||
# Add separator row (except after last entry)
|
||||
if idx < len(report_df) - 1:
|
||||
# Check if next entry is different project
|
||||
next_project = report_df.iloc[idx + 1]['專案名稱'] if idx + 1 < len(report_df) else None
|
||||
if next_project != project_name:
|
||||
# Merge cells for current project before adding separator
|
||||
if project_start_row is not None and current_row > project_start_row:
|
||||
try:
|
||||
ws.merge_cells(f'B{project_start_row}:B{current_row - 1}')
|
||||
merged_cell = ws.cell(row=project_start_row, column=2)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Merge cells for current sub-project before adding separator
|
||||
if (current_sub_project is not None and current_sub_project != "-" and
|
||||
sub_project_start_row is not None and current_row > sub_project_start_row):
|
||||
try:
|
||||
# Merge C column (sub-project name)
|
||||
ws.merge_cells(f'C{sub_project_start_row}:C{current_row - 1}')
|
||||
merged_cell = ws.cell(row=sub_project_start_row, column=3)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
|
||||
# Merge D column (progress) as well
|
||||
ws.merge_cells(f'D{sub_project_start_row}:D{current_row - 1}')
|
||||
merged_cell_d = ws.cell(row=sub_project_start_row, column=4)
|
||||
merged_cell_d.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Add separator row
|
||||
ws.cell(row=current_row, column=2, value='專案名稱')
|
||||
for col_idx, header in enumerate(headers[1:], start=3):
|
||||
ws.cell(row=current_row, column=col_idx, value=header)
|
||||
# Apply blue header styling to separator row
|
||||
style_header_row(ws, row_idx=current_row, template_cell=template_header_cell)
|
||||
current_row += 1
|
||||
|
||||
# Reset project tracking for next group
|
||||
current_project = None
|
||||
project_start_row = None
|
||||
current_sub_project = None
|
||||
sub_project_start_row = None
|
||||
|
||||
# Handle the last project merging (if needed)
|
||||
if current_project is not None and project_start_row is not None and current_row > project_start_row:
|
||||
try:
|
||||
ws.merge_cells(f'B{project_start_row}:B{current_row - 1}')
|
||||
merged_cell = ws.cell(row=project_start_row, column=2)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Handle the last sub-project merging (if needed)
|
||||
if (current_sub_project is not None and current_sub_project != "-" and
|
||||
sub_project_start_row is not None and current_row > sub_project_start_row):
|
||||
try:
|
||||
# Merge C column (sub-project name)
|
||||
ws.merge_cells(f'C{sub_project_start_row}:C{current_row - 1}')
|
||||
merged_cell = ws.cell(row=sub_project_start_row, column=3)
|
||||
merged_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
|
||||
# Merge D column (progress) as well
|
||||
ws.merge_cells(f'D{sub_project_start_row}:D{current_row - 1}')
|
||||
merged_cell_d = ws.cell(row=sub_project_start_row, column=4)
|
||||
merged_cell_d.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Apply formatting (only if no template was used)
|
||||
if not template_sheet:
|
||||
# Format title rows
|
||||
ws['B1'].font = Font(bold=True, size=14)
|
||||
ws['B1'].alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
ws['B2'].font = Font(bold=True, size=12)
|
||||
ws['B2'].alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||||
ws['B3'].font = Font(size=10, italic=True)
|
||||
ws['B3'].alignment = Alignment(vertical='top', wrap_text=True)
|
||||
|
||||
# Merge cells for headers
|
||||
ws.merge_cells('B1:J1')
|
||||
ws.merge_cells('B2:J2')
|
||||
ws.merge_cells('B3:J3')
|
||||
|
||||
# Set column widths
|
||||
column_widths = {
|
||||
'B': 20, # 專案名稱
|
||||
'C': 20, # 子項目名稱
|
||||
'D': 8, # 進度
|
||||
'E': 35, # 本周主要進展
|
||||
'F': 15, # 參與人員
|
||||
'G': 12, # 工時
|
||||
'H': 15, # 交付物
|
||||
'I': 10, # 代碼上傳
|
||||
'J': 20 # 下周計畫
|
||||
}
|
||||
|
||||
for col, width in column_widths.items():
|
||||
ws.column_dimensions[col].width = width
|
||||
|
||||
# Apply wrap text to all data cells
|
||||
wrap_align = Alignment(vertical='top', wrap_text=True)
|
||||
for row in range(5, current_row):
|
||||
for col in range(2, 11):
|
||||
ws.cell(row=row, column=col).alignment = wrap_align
|
||||
|
||||
# Save workbook
|
||||
wb.save(output_file)
|
||||
print(f"\n✓ Weekly report saved to: {output_file}")
|
||||
|
||||
return output_file, report_df
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
import sys
|
||||
|
||||
# Default paths
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
if len(sys.argv) > 3:
|
||||
# Date range provided: start_date end_date output_file
|
||||
start_date = sys.argv[1]
|
||||
end_date = sys.argv[2]
|
||||
output_file = sys.argv[3]
|
||||
json_input = (start_date, end_date)
|
||||
elif len(sys.argv) > 2:
|
||||
# JSON file provided: json_file output_file
|
||||
json_input = sys.argv[1]
|
||||
output_file = sys.argv[2]
|
||||
else:
|
||||
# Default: use dates and auto-generate output file
|
||||
from datetime import datetime, timedelta
|
||||
today = datetime.now()
|
||||
days_since_monday = today.weekday()
|
||||
this_monday = today - timedelta(days=days_since_monday)
|
||||
this_friday = this_monday + timedelta(days=4)
|
||||
|
||||
start_date = this_monday.strftime("%Y-%m-%d")
|
||||
end_date = this_friday.strftime("%Y-%m-%d")
|
||||
json_input = (start_date, end_date)
|
||||
|
||||
timestamp = datetime.now().strftime('%Y%m%d')
|
||||
output_file = os.path.join(script_dir, 'references', f'項目週報-智能控制組-{timestamp}.xlsx')
|
||||
|
||||
# Find template file
|
||||
template_file = os.path.join(script_dir, 'references', '項目週報-模板.xlsx')
|
||||
|
||||
# Generate the report
|
||||
try:
|
||||
output_path, summary_data = generate_weekly_report_from_json(
|
||||
json_input=json_input,
|
||||
output_file=output_file,
|
||||
template_file=template_file if os.path.exists(template_file) else None
|
||||
)
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("SUMMARY OF PROJECTS:")
|
||||
print("="*60)
|
||||
for idx, row in summary_data.iterrows():
|
||||
print(f"\nProject: {row['專案名稱']}")
|
||||
print(f" Sub-project: {row['子項目名稱']}")
|
||||
print(f" Team: {row['參與人員']}")
|
||||
print(f" Hours: {row['工時']}")
|
||||
if row['本周主要進展']:
|
||||
print(f" Progress:\n {row['本周主要進展'].replace(chr(10), chr(10) + ' ')}")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("✓ Report generation completed successfully!")
|
||||
print("="*60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n✗ Error generating report: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1,235 +0,0 @@
|
|||
# Week Work Package Log - 樹狀格式顯示
|
||||
|
||||
這個 skill 用於從 OpenProject API 獲取指定日期區間的工時紀錄,並以樹狀格式顯示項目和工作包的階層關係。
|
||||
|
||||
## 功能特色
|
||||
|
||||
- 📅 **日期範圍篩選**: 指定開始和結束日期來獲取工時紀錄
|
||||
- 🌳 **樹狀格式顯示**:
|
||||
- P 表示 Project (項目),顯示完整項目名稱
|
||||
- W 表示 Work Package (工作包),顯示完整工作包名稱
|
||||
- 用縮排顯示父子階層關係
|
||||
- 📊 **工時統計**: 顯示每個節點的總工時(包含子節點)
|
||||
- 👥 **詳細資訊**: 顯示用戶、留言、工時等詳細資訊
|
||||
- 🔍 **智能排序**: 按總工時從高到低排序
|
||||
- 📄 **多輸出格式**: 支援樹狀顯示、JSON格式,或同時輸出兩種格式
|
||||
- 💬 **智能留言顯示**: 優先顯示留言,無留言時顯示 "no-comment"
|
||||
- ✅ **完成度顯示**: 顯示每個work package的完成度資訊(支援遞歸計算)
|
||||
- 若關閉或駁回狀態 → 顯示 100%
|
||||
- 若有設定完成度 → 顯示該完成度
|
||||
- 若無設定完成度(no-info):
|
||||
- 無子項目(最終層)→ 顯示 0%
|
||||
- 有子項目 → 計算所有子項目完成度的平均值
|
||||
- 範例:父work package包含三個子work package完成度分別為 0%, 50%, 100%,則父work package完成度為 50%
|
||||
- 📁 **Project容器**: Project作為容器顯示,提供工時統計但不顯示完成度(完成度概念僅適用於work package)
|
||||
|
||||
## 顯示格式範例
|
||||
|
||||
```
|
||||
* P56 - production process system (24.50h) # Project作為容器,不顯示完成度
|
||||
* W756 - Templated object code scan and render then print (16.50h) [完成度: 80%] # 由子項目平均計算
|
||||
* W824 - code page logic implement (16.50h) [完成度: 100%]
|
||||
* W877 - printing & set label size (8.50h) [完成度: 100%]
|
||||
📝 2026-01-22 | 庭宇 施 | 完成列印功能 | 6.00h
|
||||
📝 2026-01-21 | 庭宇 施 | no-comment | 2.00h
|
||||
* W841 - Flexible variable type settings (4.00h) [完成度: 60%] # 由API設定的實際完成度
|
||||
📝 2026-01-22 | 庭宇 施 | no-comment | 4.00h
|
||||
* W999 - New feature (0.00h) [完成度: 0%] # 最終層無設定完成度,顯示0%
|
||||
```
|
||||
|
||||
## 使用方式
|
||||
|
||||
### 基本用法
|
||||
|
||||
```bash
|
||||
# 獲取指定日期範圍的工時紀錄(樹狀格式,預設)
|
||||
uv run --env-file .env python work_package_logger.py 2026-01-20 2026-01-24
|
||||
|
||||
# 輸出 JSON 格式
|
||||
uv run --env-file .env python work_package_logger.py 2026-01-20 2026-01-24 --format json
|
||||
|
||||
# 同時輸出樹狀格式和 JSON 格式
|
||||
uv run --env-file .env python work_package_logger.py 2026-01-20 2026-01-24 --format both
|
||||
|
||||
# 測試 API 連接
|
||||
uv run --env-file .env python work_package_logger.py 2026-01-20 2026-01-24 --test-connection
|
||||
```
|
||||
|
||||
### 在其他腳本中使用
|
||||
|
||||
```python
|
||||
from work_package_logger import WeekWorkPackageLogger
|
||||
|
||||
# 創建實例
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
# 測試連接
|
||||
if logger.test_connection():
|
||||
# 獲取並顯示樹狀格式工時紀錄
|
||||
logger.fetch_and_display_time_entries("2026-01-20", "2026-01-24")
|
||||
|
||||
# 獲取 JSON 格式資料
|
||||
json_data = logger.fetch_and_display_time_entries(
|
||||
"2026-01-20", "2026-01-24",
|
||||
output_format="json"
|
||||
)
|
||||
|
||||
# 同時輸出兩種格式
|
||||
logger.fetch_and_display_time_entries(
|
||||
"2026-01-20", "2026-01-24",
|
||||
output_format="both"
|
||||
)
|
||||
```
|
||||
|
||||
## 環境變數
|
||||
|
||||
確保設置以下環境變數:
|
||||
|
||||
```bash
|
||||
OPENPROJECT_URL=http://ixd.openproject.techmation.com.tw
|
||||
OPENPROJECT_TOKEN=your_api_token_here
|
||||
```
|
||||
|
||||
或使用 .env 檔案:
|
||||
|
||||
```
|
||||
OPENPROJECT_URL=http://ixd.openproject.techmation.com.tw
|
||||
OPENPROJECT_TOKEN=your_api_token_here
|
||||
```
|
||||
|
||||
## 輸出格式說明
|
||||
|
||||
### 1. 統計摘要
|
||||
- 總工時統計
|
||||
|
||||
### 2. 樹狀階層結構
|
||||
- **P{項目ID} - {項目名稱}**: 項目節點,顯示完整項目名稱和總工時(包含所有子節點)
|
||||
- **W{工作包ID} - {工作包名稱}**: 工作包節點,顯示完整工作包名稱和總工時
|
||||
- **縮排**: 表示父子關係,越深層的節點縮排越多
|
||||
- **工時標示**: `(X.XXh)` 顯示該節點的總工時
|
||||
|
||||
### 3. 時間條目詳細資訊
|
||||
對於有實際時間記錄的工作包,會顯示:
|
||||
- 📝 日期 | 用戶名稱 | 留言內容(或 "no-comment")| 工時
|
||||
- 優先顯示留言內容,若留言為空則顯示 "no-comment"
|
||||
|
||||
### 4. JSON 輸出格式
|
||||
當使用 `--format json` 時,輸出結構化 JSON 資料:
|
||||
```json
|
||||
{
|
||||
"summary": {
|
||||
"date_range": "2026-01-22 to 2026-01-22",
|
||||
"total_hours": 24.0,
|
||||
"total_projects": 3,
|
||||
"generation_time": "2026-01-23T16:13:33.246620"
|
||||
},
|
||||
"projects": [
|
||||
{
|
||||
"id": "56",
|
||||
"name": "production process system",
|
||||
"type": "project",
|
||||
"hours": 0.0,
|
||||
"total_hours": 8.0,
|
||||
"time_entries": [],
|
||||
"children": [...]
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### 4. 排序邏輯
|
||||
- 同層級的節點按總工時從高到低排序
|
||||
- 確保最重要的項目和工作包優先顯示
|
||||
|
||||
## 階層關係說明
|
||||
|
||||
樹狀結構反映了實際的項目組織架構:
|
||||
|
||||
1. **項目階層**:
|
||||
- 父項目 → 子項目 → 孫項目...
|
||||
|
||||
2. **工作包階層**:
|
||||
- 父任務 → 子任務 → 子子任務...
|
||||
|
||||
3. **從屬關係**:
|
||||
- 工作包屬於項目
|
||||
- 子工作包屬於父工作包
|
||||
|
||||
## 錯誤處理
|
||||
|
||||
- ✅ API 連接測試
|
||||
- ✅ 日期格式驗證 (YYYY-MM-DD)
|
||||
- ✅ 日期範圍邏輯檢查
|
||||
- ✅ 大日期範圍警告 (>90天)
|
||||
- ✅ 分頁數據獲取
|
||||
- ✅ 異常時間條目跳過並警告
|
||||
- ✅ 階層循環檢測和深度限制
|
||||
|
||||
## 依賴套件
|
||||
|
||||
- `requests`: HTTP API 請求
|
||||
- `datetime`: 日期處理
|
||||
- `json`: JSON 數據處理
|
||||
- `re`: 正則表達式(時間格式解析)
|
||||
- `collections`: 數據結構(樹狀結構建構)
|
||||
|
||||
## 限制
|
||||
|
||||
- 日期格式必須為 YYYY-MM-DD
|
||||
- 需要有效的 OpenProject API token
|
||||
- 大量數據獲取可能需要較長時間
|
||||
- 依賴 OpenProject API v3 格式
|
||||
- 階層深度限制為 10 層(防止無限循環)
|
||||
|
||||
## 進階功能
|
||||
|
||||
- **智能緩存**: 自動緩存工作包和項目資訊,避免重複 API 調用
|
||||
- **容錯處理**: 即使部分數據獲取失敗,仍會顯示可用的資訊
|
||||
- **靈活顯示**: 支援任意深度的階層結構顯示
|
||||
- **多重輸出格式**:
|
||||
- **樹狀格式** (預設): 適合人類閱讀的階層結構顯示
|
||||
- **JSON格式**: 適合機器處理和資料轉換的結構化格式
|
||||
- **同時輸出**: 可同時提供兩種格式輸出
|
||||
- **智能留言顯示**: 優先顯示時間條目的留言內容,無留言時顯示 "no-comment"
|
||||
- **完整名稱顯示**: 項目和工作包都顯示完整名稱,不只是ID
|
||||
- **階層智能合併**: 自動合併並顯示完整的項目和工作包階層關係
|
||||
|
||||
## 輸出格式選項
|
||||
|
||||
### 1. 樹狀格式 (預設)
|
||||
```bash
|
||||
--format tree
|
||||
```
|
||||
- 適合人類閱讀
|
||||
- 清晰的階層結構
|
||||
- 工時統計和時間條目詳細資訊
|
||||
|
||||
### 2. JSON格式
|
||||
```bash
|
||||
--format json
|
||||
```
|
||||
- 完整的結構化資料
|
||||
- 適合程式處理
|
||||
- 包含摘要資訊和完整階層
|
||||
- 方便轉換為其他格式(如Excel)
|
||||
|
||||
### 3. 同時輸出
|
||||
```bash
|
||||
--format both
|
||||
```
|
||||
- 先顯示樹狀格式
|
||||
- 接著輸出JSON格式
|
||||
- 兼顧可讀性和資料處理需求
|
||||
|
||||
## 貢獻者與致謝
|
||||
|
||||
本 skill 的開發受益於以下貢獻:
|
||||
|
||||
### 主要貢獻
|
||||
- **kinoshitakenta** - 來自 [openproject-work-time-reports](https://ixd.gitea.techmation.com.tw/Zhubei_iXD/openproject-work-time-reports) 專案的重要貢獻
|
||||
|
||||
### 致謝
|
||||
感謝所有為 OpenProject API 整合和工時報告功能做出貢獻的開發者們。
|
||||
|
||||
---
|
||||
*Built with ❤️ for iXD Team*
|
||||
|
|
@ -1,249 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Week Work Package Log 使用範例腳本
|
||||
展示樹狀格式顯示的各種使用場景
|
||||
|
||||
Credits:
|
||||
- kinoshitakenta: Based on contributions from openproject-work-time-reports
|
||||
(https://ixd.gitea.techmation.com.tw/Zhubei_iXD/openproject-work-time-reports)
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Add the skill to Python path
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, script_dir)
|
||||
|
||||
from work_package_logger import WeekWorkPackageLogger
|
||||
|
||||
|
||||
def example_this_week():
|
||||
"""範例1: 獲取本週工時紀錄"""
|
||||
|
||||
today = datetime.now()
|
||||
days_since_monday = today.weekday()
|
||||
this_monday = today - timedelta(days=days_since_monday)
|
||||
this_friday = this_monday + timedelta(days=4)
|
||||
|
||||
start_date = this_monday.strftime("%Y-%m-%d")
|
||||
end_date = this_friday.strftime("%Y-%m-%d")
|
||||
|
||||
print(f"📅 範例1: 本週工時紀錄樹狀顯示 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
logger.fetch_and_display_time_entries(start_date, end_date)
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_specific_dates():
|
||||
"""範例2: 獲取指定日期範圍的工時紀錄"""
|
||||
|
||||
start_date = "2026-01-20"
|
||||
end_date = "2026-01-22"
|
||||
|
||||
print(f"📅 範例2: 指定日期範圍樹狀顯示 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
# 先測試連接
|
||||
print("🔗 測試 API 連接...")
|
||||
if not logger.test_connection():
|
||||
print("❌ 無法連接到 API")
|
||||
return
|
||||
|
||||
print()
|
||||
logger.fetch_and_display_time_entries(start_date, end_date)
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_single_day():
|
||||
"""範例3: 獲取單日工時紀錄"""
|
||||
|
||||
target_date = "2026-01-21"
|
||||
|
||||
print(f"📅 範例3: 單日工時紀錄樹狀顯示 ({target_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
logger.fetch_and_display_time_entries(target_date, target_date)
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_connection_test():
|
||||
"""範例4: 僅測試 API 連接"""
|
||||
|
||||
print("📅 範例4: API 連接測試")
|
||||
print("=" * 30)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
print("🔗 測試 OpenProject API 連接...")
|
||||
if logger.test_connection():
|
||||
print("✅ API 連接成功!")
|
||||
print("📋 系統功能:")
|
||||
print(" - ✅ 項目階層支援已啟用")
|
||||
print(" - ✅ 工作包階層檢測")
|
||||
print(" - ✅ 樹狀格式顯示")
|
||||
print(" - ✅ JSON 格式輸出")
|
||||
print(" - ✅ 智能留言顯示")
|
||||
else:
|
||||
print("❌ API 連接失敗")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_json_output():
|
||||
"""範例5: JSON 格式輸出"""
|
||||
|
||||
start_date = "2026-01-22"
|
||||
end_date = "2026-01-22"
|
||||
|
||||
print(f"📅 範例5: JSON 格式輸出 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
print("🔗 測試 API 連接...")
|
||||
if not logger.test_connection():
|
||||
print("❌ 無法連接到 API")
|
||||
return
|
||||
|
||||
print("\n📄 JSON 格式輸出:")
|
||||
print("-" * 50)
|
||||
json_data = logger.fetch_and_display_time_entries(start_date, end_date, 'json')
|
||||
|
||||
if json_data:
|
||||
print(f"\n📊 總結:")
|
||||
print(f" - 總工時:{json_data['summary']['total_hours']} 小時")
|
||||
print(f" - 專案數量:{json_data['summary']['total_projects']} 個")
|
||||
print(f" - 日期範圍:{json_data['summary']['date_range']}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_both_formats():
|
||||
"""範例6: 同時輸出樹狀和 JSON 格式"""
|
||||
|
||||
start_date = "2026-01-22"
|
||||
end_date = "2026-01-22"
|
||||
|
||||
print(f"📅 範例6: 同時輸出兩種格式 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
print("🔗 測試 API 連接...")
|
||||
if not logger.test_connection():
|
||||
print("❌ 無法連接到 API")
|
||||
return
|
||||
|
||||
print("\n🌳 + 📄 樹狀格式 + JSON 格式:")
|
||||
print("-" * 50)
|
||||
logger.fetch_and_display_time_entries(start_date, end_date, 'both')
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def example_completion_feature():
|
||||
"""範例7: 完成度功能展示"""
|
||||
|
||||
start_date = "2026-01-22"
|
||||
end_date = "2026-01-22"
|
||||
|
||||
print(f"📅 範例7: 遞歸完成度功能展示 ({start_date} ~ {end_date})")
|
||||
print("=" * 70)
|
||||
print("此功能會顯示每個project與work_package的完成度:")
|
||||
print(" 1. 若關閉或駁回 → 100%")
|
||||
print(" 2. 若有設定完成度 → 顯示該完成度")
|
||||
print(" 3. 若無設定(no-info):")
|
||||
print(" - 無子項目(最終層)→ 0%")
|
||||
print(" - 有子項目 → 計算所有子項目完成度平均值")
|
||||
print(" 4. 遞歸應用上述規則到所有層級")
|
||||
print("-" * 50)
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
print("🔗 測試 API 連接...")
|
||||
if not logger.test_connection():
|
||||
print("❌ 無法連接到 API")
|
||||
return
|
||||
|
||||
print("\n🌳 樹狀格式顯示(包含遞歸計算完成度):")
|
||||
logger.fetch_and_display_time_entries(start_date, end_date)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
|
||||
|
||||
def main():
|
||||
"""主函數 - 展示所有範例"""
|
||||
|
||||
print("🚀 Week Work Package Log 樹狀格式顯示範例")
|
||||
print("=" * 50)
|
||||
print()
|
||||
|
||||
while True:
|
||||
print("請選擇要執行的範例:")
|
||||
print("1. 本週工時記錄 (樹狀顯示)")
|
||||
print("2. 指定日期範圍 (2026-01-20 ~ 2026-01-22)")
|
||||
print("3. 單日工時記錄 (2026-01-21)")
|
||||
print("4. API 連接測試")
|
||||
print("5. JSON 格式輸出")
|
||||
print("6. 同時輸出兩種格式 (樹狀 + JSON)")
|
||||
print("7. 完成度功能展示")
|
||||
print("0. 退出")
|
||||
print()
|
||||
|
||||
try:
|
||||
choice = input("請輸入選項 (0-7): ").strip()
|
||||
|
||||
if choice == "0":
|
||||
print("👋 再見!")
|
||||
break
|
||||
elif choice == "1":
|
||||
example_this_week()
|
||||
elif choice == "2":
|
||||
example_specific_dates()
|
||||
elif choice == "3":
|
||||
example_single_day()
|
||||
elif choice == "4":
|
||||
example_connection_test()
|
||||
elif choice == "5":
|
||||
example_json_output()
|
||||
elif choice == "6":
|
||||
example_both_formats()
|
||||
elif choice == "7":
|
||||
example_completion_feature()
|
||||
else:
|
||||
print("❌ 無效選項,請重新輸入")
|
||||
|
||||
input("\n按 Enter 鍵繼續...")
|
||||
print("\n" + "=" * 70 + "\n")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n👋 程式已中斷,再見!")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"❌ 錯誤: {e}")
|
||||
input("\n按 Enter 鍵繼續...")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -1,902 +0,0 @@
|
|||
"""
|
||||
Week Work Package Log - OpenProject API client with tree-style display
|
||||
|
||||
Fetches time entries for work packages and displays them in a tree format
|
||||
showing the hierarchical relationships between projects and work packages.
|
||||
|
||||
Contributors:
|
||||
- kinoshitakenta: Original contributions from openproject-work-time-reports
|
||||
(https://ixd.gitea.techmation.com.tw/Zhubei_iXD/openproject-work-time-reports)
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import requests
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Union, Tuple
|
||||
import re
|
||||
from requests.auth import HTTPBasicAuth
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
|
||||
# Add project_hierarchy skill to path for project hierarchy functionality
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
skills_dir = os.path.dirname(script_dir)
|
||||
project_hierarchy_path = os.path.join(skills_dir, 'project_hierarchy')
|
||||
if project_hierarchy_path not in sys.path:
|
||||
sys.path.insert(0, project_hierarchy_path)
|
||||
|
||||
try:
|
||||
from project_manager import ProjectManager
|
||||
PROJECT_HIERARCHY_AVAILABLE = True
|
||||
except ImportError:
|
||||
PROJECT_HIERARCHY_AVAILABLE = False
|
||||
|
||||
|
||||
# Completion status constants
|
||||
COMPLETION_100_PERCENT = "100%"
|
||||
COMPLETION_0_PERCENT = "0%"
|
||||
NO_INFO = "no-info"
|
||||
PERCENT_FORMAT = "{:.0f}%"
|
||||
|
||||
# Status rejection keywords
|
||||
REJECTION_KEYWORDS = ['reject', 'denied', 'declined', '駁回', '拒絕']
|
||||
|
||||
# Completion detection patterns
|
||||
COMPLETION_KEYWORDS = ['100%', '完成', 'completed', 'finished']
|
||||
|
||||
|
||||
class WorkPackageLogError(Exception):
|
||||
"""Base exception for work package logging errors"""
|
||||
pass
|
||||
|
||||
class APIConnectionError(WorkPackageLogError):
|
||||
"""Raised when API connection fails"""
|
||||
pass
|
||||
|
||||
class AuthenticationError(WorkPackageLogError):
|
||||
"""Raised when API authentication fails"""
|
||||
pass
|
||||
|
||||
class InvalidDateRangeError(WorkPackageLogError):
|
||||
"""Raised when date range is invalid"""
|
||||
pass
|
||||
|
||||
|
||||
class TreeNode:
|
||||
"""Represents a node in the hierarchy tree"""
|
||||
def __init__(self, node_id: str, name: str, node_type: str, hours: float = 0.0, entries: List = None, completion: str = None, logger = None):
|
||||
self.id = node_id
|
||||
self.name = name
|
||||
self.type = node_type # 'project' or 'work_package'
|
||||
self.hours = hours
|
||||
self.entries = entries or []
|
||||
self.children = []
|
||||
self.parent = None
|
||||
self._raw_completion = completion # API原始完成度資料
|
||||
self._calculated_completion = None # 快取計算後的完成度
|
||||
self.logger = logger # 用於檢查駁回狀態
|
||||
|
||||
def add_child(self, child):
|
||||
child.parent = self
|
||||
self.children.append(child)
|
||||
|
||||
def get_total_hours(self) -> float:
|
||||
"""Get total hours including all children"""
|
||||
total = self.hours
|
||||
for child in self.children:
|
||||
total += child.get_total_hours()
|
||||
return total
|
||||
|
||||
def _is_rejected(self) -> bool:
|
||||
"""檢查此節點是否為駁回狀態"""
|
||||
if self.type != 'work_package' or not self.logger:
|
||||
return False
|
||||
|
||||
return self.logger._is_work_package_rejected(self.id)
|
||||
|
||||
def get_completion(self, visited_nodes: set = None) -> str:
|
||||
"""
|
||||
Recursively compute the completion percentage for this node.
|
||||
|
||||
The calculation follows these rules:
|
||||
- If a cached value exists in ``_calculated_completion``, return it immediately.
|
||||
- If the raw completion (``_raw_completion``) is exactly ``"100%"``,
|
||||
this node is treated as fully complete and returns ``"100%"``.
|
||||
- If ``_raw_completion`` is set and not equal to ``"no-info"``,
|
||||
that value is returned as-is.
|
||||
- If there is no meaningful raw completion and the node has no children
|
||||
(a leaf in the hierarchy), the completion is treated as ``"0%"``.
|
||||
- Otherwise, the completion is derived from the children (for example,
|
||||
by aggregating their completion values according to the hierarchy).
|
||||
|
||||
To prevent infinite recursion in case of cycles in the hierarchy graph,
|
||||
the method tracks visited node IDs in ``visited_nodes``. If the current
|
||||
node's ID is already present in ``visited_nodes``, the method returns
|
||||
``"no-info"`` to indicate that a reliable completion value cannot be
|
||||
determined.
|
||||
|
||||
The computed completion value is cached in ``_calculated_completion`` so
|
||||
that subsequent calls do not need to traverse the hierarchy again.
|
||||
|
||||
Args:
|
||||
visited_nodes: Optional set used internally during recursive calls
|
||||
to record which node IDs have already been visited. Callers
|
||||
should normally omit this argument and allow the method to
|
||||
create and manage the set.
|
||||
|
||||
Returns:
|
||||
A string representing the completion percentage for this node
|
||||
(for example, ``"0%"``, ``"50%"``, ``"100%"`` or ``"no-info"``).
|
||||
"""
|
||||
if visited_nodes is None:
|
||||
visited_nodes = set()
|
||||
|
||||
# 避免無限遞歸
|
||||
if self.id in visited_nodes:
|
||||
return "no-info"
|
||||
visited_nodes.add(self.id)
|
||||
|
||||
# 如果已經計算過,直接返回快取
|
||||
if self._calculated_completion is not None:
|
||||
return self._calculated_completion
|
||||
|
||||
# 1. 檢查是否為關閉或駁回狀態
|
||||
if self._raw_completion == "100%":
|
||||
self._calculated_completion = "100%"
|
||||
return "100%"
|
||||
|
||||
# 2. 如果有子項目,需要重新計算(跳過駁回的子項目)
|
||||
if self.children:
|
||||
# 有子項目,計算所有子項目完成度的平均值(跳過駁回的子項目)
|
||||
child_completions = []
|
||||
for child in self.children:
|
||||
# 跳過駁回狀態的子項目
|
||||
if child._is_rejected():
|
||||
continue
|
||||
|
||||
child_completion_str = child.get_completion(visited_nodes)
|
||||
|
||||
# 將完成度字串轉換為數值
|
||||
if child_completion_str == "no-info":
|
||||
child_completions.append(0)
|
||||
elif child_completion_str.endswith("%"):
|
||||
try:
|
||||
child_completions.append(float(child_completion_str[:-1]))
|
||||
except ValueError:
|
||||
child_completions.append(0)
|
||||
else:
|
||||
child_completions.append(0)
|
||||
|
||||
if child_completions:
|
||||
avg_completion = sum(child_completions) / len(child_completions)
|
||||
self._calculated_completion = f"{avg_completion:.0f}%"
|
||||
return self._calculated_completion
|
||||
# 如果所有子項目都被駁回,則使用原始完成度或0%
|
||||
|
||||
# 3. 檢查是否有設定完成度(沒有子項目或所有子項目都被駁回時使用)
|
||||
# 如果設定為 0%,視為未設定,仍會嘗試從子項目計算
|
||||
if self._raw_completion and self._raw_completion != "no-info" and self._raw_completion != "0%":
|
||||
self._calculated_completion = self._raw_completion
|
||||
return self._raw_completion
|
||||
|
||||
# 4. 沒有任何完成度資訊,給0%
|
||||
self._calculated_completion = "0%"
|
||||
return "0%"
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
"""Convert tree node to dictionary for JSON serialization"""
|
||||
result = {
|
||||
'id': self.id,
|
||||
'name': self.name,
|
||||
'type': self.type,
|
||||
'hours': self.hours,
|
||||
'total_hours': self.get_total_hours(),
|
||||
'time_entries': self.entries,
|
||||
'children': [child.to_dict() for child in sorted(self.children, key=lambda x: x.get_total_hours(), reverse=True)]
|
||||
}
|
||||
|
||||
# 只有work package才顯示完成度
|
||||
if self.type == 'work_package':
|
||||
result['completion'] = self.get_completion()
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class WeekWorkPackageLogger:
|
||||
"""
|
||||
Fetches and displays time entries for work packages in a tree format.
|
||||
|
||||
Features:
|
||||
- Tree-style display showing project and work package hierarchies
|
||||
- Date range filtering for time entries
|
||||
- Detailed time entry information display
|
||||
- User and project information extraction
|
||||
"""
|
||||
|
||||
def __init__(self, api_url: str = None, api_token: str = None):
|
||||
"""
|
||||
Initialize WeekWorkPackageLogger.
|
||||
|
||||
Args:
|
||||
api_url: OpenProject base URL
|
||||
api_token: OpenProject API token
|
||||
"""
|
||||
self.api_url = (api_url or os.getenv("OPENPROJECT_URL", "http://ixd.openproject.techmation.com.tw")).rstrip('/')
|
||||
if self.api_url.endswith('/api/v3'):
|
||||
self.api_url = self.api_url[:-7]
|
||||
|
||||
self.api_token = api_token or os.getenv("OPENPROJECT_TOKEN")
|
||||
|
||||
# API endpoints
|
||||
self.time_entries_endpoint = f"{self.api_url}/api/v3/time_entries"
|
||||
self.users_endpoint = f"{self.api_url}/api/v3/users"
|
||||
self.projects_endpoint = f"{self.api_url}/api/v3/projects"
|
||||
self.work_packages_endpoint = f"{self.api_url}/api/v3/work_packages"
|
||||
|
||||
# Initialize project manager for hierarchy support
|
||||
self.project_manager = None
|
||||
if PROJECT_HIERARCHY_AVAILABLE:
|
||||
try:
|
||||
self.project_manager = ProjectManager(
|
||||
api_url=self.api_url,
|
||||
api_token=self.api_token
|
||||
)
|
||||
print("✓ Project hierarchy support enabled")
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not initialize project hierarchy: {e}")
|
||||
self.project_manager = None
|
||||
|
||||
# Cache for work package and project data
|
||||
self.work_package_cache = {}
|
||||
self.project_cache = {}
|
||||
self.status_cache = {}
|
||||
|
||||
def _get_auth(self):
|
||||
"""Return authentication tuple for API requests"""
|
||||
if not self.api_token:
|
||||
raise AuthenticationError("No API token provided. Set OPENPROJECT_TOKEN environment variable or pass api_token parameter.")
|
||||
return ("apikey", self.api_token)
|
||||
|
||||
def test_connection(self) -> bool:
|
||||
"""Test API connection and authentication"""
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{self.api_url}/api/v3",
|
||||
auth=self._get_auth(),
|
||||
timeout=10
|
||||
)
|
||||
response.raise_for_status()
|
||||
print("✓ API connection successful")
|
||||
return True
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"✗ API connection failed: {e}")
|
||||
return False
|
||||
|
||||
def _validate_date_range(self, start_date: str, end_date: str):
|
||||
"""Validate and parse date range"""
|
||||
try:
|
||||
start_dt = datetime.strptime(start_date, "%Y-%m-%d")
|
||||
end_dt = datetime.strptime(end_date, "%Y-%m-%d")
|
||||
|
||||
if start_dt > end_dt:
|
||||
raise InvalidDateRangeError("Start date must be before or equal to end date")
|
||||
|
||||
if (end_dt - start_dt).days > 90:
|
||||
print("⚠ Warning: Large date range may take longer to fetch")
|
||||
|
||||
return start_dt, end_dt
|
||||
|
||||
except ValueError as e:
|
||||
raise InvalidDateRangeError(f"Invalid date format. Use YYYY-MM-DD format: {e}")
|
||||
|
||||
def _parse_duration(self, duration_str: str) -> float:
|
||||
"""Parse ISO 8601 duration format (PT8H30M) to hours"""
|
||||
if not duration_str:
|
||||
return 0.0
|
||||
|
||||
# Handle ISO 8601 format like PT8H30M
|
||||
match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?', duration_str)
|
||||
if match:
|
||||
hours = int(match.group(1) or 0)
|
||||
minutes = int(match.group(2) or 0)
|
||||
return hours + (minutes / 60.0)
|
||||
|
||||
# Handle simple numeric format
|
||||
try:
|
||||
return float(duration_str)
|
||||
except ValueError:
|
||||
print(f"⚠ Warning: Could not parse duration '{duration_str}', using 0")
|
||||
return 0.0
|
||||
|
||||
def _build_filter_params(self, start_date: str, end_date: str) -> Dict:
|
||||
"""Build filter parameters for API request"""
|
||||
filters = []
|
||||
|
||||
# Date range filter
|
||||
filters.append({
|
||||
"spentOn": {
|
||||
"operator": "<>d",
|
||||
"values": [start_date, end_date]
|
||||
}
|
||||
})
|
||||
|
||||
return {"filters": json.dumps(filters)}
|
||||
|
||||
def _fetch_paginated_data(self, url: str, params: Dict = None) -> List[Dict]:
|
||||
"""Fetch all pages of data from a paginated API endpoint"""
|
||||
all_data = []
|
||||
offset = 0
|
||||
page_size = 100
|
||||
|
||||
params = params or {}
|
||||
|
||||
while True:
|
||||
current_params = {
|
||||
**params,
|
||||
"offset": offset,
|
||||
"pageSize": page_size
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.get(
|
||||
url,
|
||||
params=current_params,
|
||||
auth=self._get_auth(),
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
elements = data.get('_embedded', {}).get('elements', [])
|
||||
if not elements:
|
||||
break
|
||||
|
||||
all_data.extend(elements)
|
||||
|
||||
total = data.get('total', 0)
|
||||
if offset + page_size >= total:
|
||||
break
|
||||
|
||||
offset += page_size
|
||||
|
||||
if len(all_data) % 500 == 0:
|
||||
print(f"📥 Fetched {len(all_data)} entries...")
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise APIConnectionError(f"Failed to fetch data from {url}: {e}")
|
||||
|
||||
return all_data
|
||||
|
||||
def _get_linked_info(self, time_entry: Dict, link_name: str, field_name: str = 'title') -> str:
|
||||
"""Extract linked information from time entry"""
|
||||
try:
|
||||
link = time_entry.get('_links', {}).get(link_name)
|
||||
if link and isinstance(link, dict):
|
||||
return link.get(field_name, '')
|
||||
return ''
|
||||
except Exception:
|
||||
return ''
|
||||
|
||||
def _get_linked_id(self, entry: Dict, link_type: str) -> str:
|
||||
"""Extract ID from linked resources in time entry"""
|
||||
try:
|
||||
links = entry.get('_links', {})
|
||||
resource = links.get(link_type, {})
|
||||
|
||||
if resource and 'href' in resource:
|
||||
href = resource['href']
|
||||
return href.split('/')[-1]
|
||||
|
||||
return ''
|
||||
except Exception:
|
||||
return ''
|
||||
|
||||
def _get_work_package_details(self, work_package_id: str) -> Dict:
|
||||
"""Get work package details from API with caching"""
|
||||
if work_package_id in self.work_package_cache:
|
||||
return self.work_package_cache[work_package_id]
|
||||
|
||||
try:
|
||||
wp_url = f"{self.work_packages_endpoint}/{work_package_id}"
|
||||
response = requests.get(wp_url, auth=self._get_auth(), timeout=10)
|
||||
|
||||
if response.status_code == 200:
|
||||
wp_data = response.json()
|
||||
self.work_package_cache[work_package_id] = wp_data
|
||||
return wp_data
|
||||
else:
|
||||
return {}
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not fetch work package {work_package_id}: {e}")
|
||||
return {}
|
||||
|
||||
def _get_status_details(self, status_href: str) -> Dict:
|
||||
"""Get status details from API with caching"""
|
||||
if status_href in self.status_cache:
|
||||
return self.status_cache[status_href]
|
||||
|
||||
try:
|
||||
response = requests.get(f"{self.api_url}{status_href}", auth=self._get_auth(), timeout=10)
|
||||
|
||||
if response.status_code == 200:
|
||||
status_data = response.json()
|
||||
self.status_cache[status_href] = status_data
|
||||
return status_data
|
||||
else:
|
||||
return {}
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not fetch status {status_href}: {e}")
|
||||
return {}
|
||||
|
||||
|
||||
def _is_work_package_rejected(self, work_package_id: str) -> bool:
|
||||
"""檢查 work package 是否為駁回狀態"""
|
||||
wp_data = self._get_work_package_details(work_package_id)
|
||||
if not wp_data:
|
||||
return False
|
||||
|
||||
# 檢查狀態是否包含駁回關鍵字
|
||||
status_link = wp_data.get('_links', {}).get('status')
|
||||
if status_link and status_link.get('href'):
|
||||
status_data = self._get_status_details(status_link['href'])
|
||||
if status_data:
|
||||
status_name = status_data.get('name', '').lower()
|
||||
if any(keyword in status_name for keyword in REJECTION_KEYWORDS):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def _get_work_package_completion(self, work_package_id: str) -> str:
|
||||
"""獲取work package的原始完成度資訊"""
|
||||
wp_data = self._get_work_package_details(work_package_id)
|
||||
if not wp_data:
|
||||
return NO_INFO
|
||||
|
||||
# 1. 檢查狀態是否為關閉或駁回
|
||||
status_link = wp_data.get('_links', {}).get('status')
|
||||
if status_link and status_link.get('href'):
|
||||
status_data = self._get_status_details(status_link['href'])
|
||||
if status_data:
|
||||
# 檢查是否為關閉狀態
|
||||
if status_data.get('isClosed', False):
|
||||
return COMPLETION_100_PERCENT
|
||||
|
||||
# 檢查狀態名稱是否包含「駁回」相關字詞
|
||||
status_name = status_data.get('name', '').lower()
|
||||
if any(keyword in status_name for keyword in REJECTION_KEYWORDS):
|
||||
return COMPLETION_100_PERCENT
|
||||
|
||||
# 2. 檢查work package本身的完成度設定
|
||||
percentage_done = wp_data.get('percentageDone')
|
||||
if percentage_done is not None:
|
||||
return f"{percentage_done}%"
|
||||
|
||||
derived_percentage = wp_data.get('derivedPercentageDone')
|
||||
if derived_percentage is not None:
|
||||
return f"{derived_percentage}%"
|
||||
|
||||
# 3. 若都沒有則填 no-info
|
||||
return NO_INFO
|
||||
|
||||
def _get_work_package_children(self, work_package_id: str) -> List[Dict]:
|
||||
"""獲取 work package 的所有直接子項目"""
|
||||
try:
|
||||
# 使用 OpenProject API 獲取子項目
|
||||
params = {
|
||||
'filters': json.dumps([{
|
||||
'parent': {
|
||||
'operator': '=',
|
||||
'values': [work_package_id]
|
||||
}
|
||||
}])
|
||||
}
|
||||
|
||||
response = requests.get(
|
||||
self.work_packages_endpoint,
|
||||
params=params,
|
||||
auth=self._get_auth(),
|
||||
timeout=30
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
return data.get('_embedded', {}).get('elements', [])
|
||||
else:
|
||||
return []
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not fetch children for work package {work_package_id}: {e}")
|
||||
return []
|
||||
|
||||
def _build_complete_work_package_tree(self, work_package_id: str, wp_nodes: Dict, time_entries_by_wp: Dict) -> TreeNode:
|
||||
"""遞歸建構完整的 work package 樹狀結構(包含所有子項目)"""
|
||||
if work_package_id in wp_nodes:
|
||||
return wp_nodes[work_package_id]
|
||||
|
||||
# 獲取 work package 詳細資料
|
||||
wp_data = self._get_work_package_details(work_package_id)
|
||||
if not wp_data:
|
||||
return None
|
||||
|
||||
wp_name = wp_data.get('subject', f'WP-{work_package_id}')
|
||||
|
||||
# 檢查是否有時間記錄
|
||||
entries = time_entries_by_wp.get(work_package_id, [])
|
||||
node_hours = sum(entry['hours'] for entry in entries)
|
||||
|
||||
# 獲取完成度
|
||||
wp_completion = self._get_work_package_completion(work_package_id)
|
||||
|
||||
# 創建節點
|
||||
node = TreeNode(work_package_id, wp_name, 'work_package', node_hours, entries, wp_completion, self)
|
||||
wp_nodes[work_package_id] = node
|
||||
|
||||
# 獲取所有子項目
|
||||
children_data = self._get_work_package_children(work_package_id)
|
||||
for child_data in children_data:
|
||||
child_id = str(child_data.get('id'))
|
||||
child_node = self._build_complete_work_package_tree(child_id, wp_nodes, time_entries_by_wp)
|
||||
if child_node:
|
||||
node.add_child(child_node)
|
||||
|
||||
return node
|
||||
|
||||
def _get_project_completion(self, project_id: str) -> str:
|
||||
"""Project不計算完成度"""
|
||||
return None
|
||||
|
||||
def _get_project_path(self, project_id: str) -> List[Tuple[str, str]]:
|
||||
"""Get full project path from root to target project"""
|
||||
if not self.project_manager or not project_id:
|
||||
return []
|
||||
|
||||
try:
|
||||
if not self.project_manager.projects:
|
||||
self.project_manager.fetch_projects()
|
||||
|
||||
# Find project identifier by ID
|
||||
project_identifier = None
|
||||
for identifier, project_data in self.project_manager.projects.items():
|
||||
if str(project_data.get('id')) == str(project_id):
|
||||
project_identifier = identifier
|
||||
break
|
||||
|
||||
if not project_identifier:
|
||||
return []
|
||||
|
||||
# Get full path identifiers
|
||||
path_identifiers = self.project_manager.get_full_path(project_identifier)
|
||||
|
||||
# Convert to (id, name) tuples
|
||||
path = []
|
||||
for path_id in path_identifiers:
|
||||
if path_id in self.project_manager.projects:
|
||||
proj_data = self.project_manager.projects[path_id]
|
||||
path.append((str(proj_data['id']), proj_data['name']))
|
||||
|
||||
return path
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not get project path for {project_id}: {e}")
|
||||
return []
|
||||
|
||||
def _get_work_package_path(self, work_package_id: str) -> List[Tuple[str, str]]:
|
||||
"""Get full work package path from root to target work package"""
|
||||
path = []
|
||||
current_id = work_package_id
|
||||
visited = set()
|
||||
max_depth = 10
|
||||
depth = 0
|
||||
|
||||
while current_id and depth < max_depth and current_id not in visited:
|
||||
visited.add(current_id)
|
||||
|
||||
wp_data = self._get_work_package_details(current_id)
|
||||
if not wp_data:
|
||||
break
|
||||
|
||||
wp_subject = wp_data.get('subject', f'WP-{current_id}')
|
||||
path.insert(0, (current_id, wp_subject))
|
||||
|
||||
# Check for parent
|
||||
links = wp_data.get('_links', {})
|
||||
parent_link = links.get('parent')
|
||||
|
||||
if parent_link and parent_link.get('href'):
|
||||
current_id = parent_link.get('href', '').split('/')[-1]
|
||||
else:
|
||||
break
|
||||
|
||||
depth += 1
|
||||
|
||||
return path
|
||||
|
||||
def _build_tree_structure(self, time_entries: List[Dict]) -> Dict[str, TreeNode]:
|
||||
"""Build tree structure from time entries"""
|
||||
# Sort time entries by ID ascending (oldest first)
|
||||
time_entries_sorted = sorted(time_entries, key=lambda x: int(x.get('id', 0)))
|
||||
|
||||
# Create root nodes for projects
|
||||
project_trees = {}
|
||||
|
||||
# Group entries by project and work package
|
||||
# Note: project_id here is the direct project associated with the time entry,
|
||||
# which may not be the root project but a child project in the hierarchy
|
||||
project_entries = defaultdict(lambda: defaultdict(list))
|
||||
|
||||
for entry in time_entries_sorted:
|
||||
try:
|
||||
# Extract basic information
|
||||
spent_on = entry.get('spentOn', 'Unknown')
|
||||
comment_raw = entry.get('comment', '')
|
||||
# Extract raw comment from comment object structure
|
||||
if isinstance(comment_raw, dict):
|
||||
comment = comment_raw.get('raw', '').strip()
|
||||
else:
|
||||
comment = '' if comment_raw is None else str(comment_raw).strip()
|
||||
hours_str = entry.get('hours', 'PT0H')
|
||||
hours = self._parse_duration(hours_str)
|
||||
|
||||
# Extract linked information
|
||||
user_name = self._get_linked_info(entry, 'user', 'title')
|
||||
project_id = self._get_linked_id(entry, 'project') # This is the actual project from time entry
|
||||
activity_name = self._get_linked_info(entry, 'activity', 'title')
|
||||
work_package_id = self._get_linked_id(entry, 'workPackage')
|
||||
|
||||
if not project_id or not work_package_id:
|
||||
continue
|
||||
|
||||
entry_data = {
|
||||
'date': spent_on,
|
||||
'user': user_name,
|
||||
'activity': activity_name,
|
||||
'comment': comment,
|
||||
'hours': hours
|
||||
}
|
||||
|
||||
# Group by the direct project from time entry
|
||||
project_entries[project_id][work_package_id].append(entry_data)
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Error processing time entry {entry.get('id', 'unknown')}: {e}")
|
||||
continue
|
||||
|
||||
# Build tree structure for each project
|
||||
for project_id, wp_entries in project_entries.items():
|
||||
# Get the full project hierarchy path from root to this project
|
||||
project_path = self._get_project_path(project_id)
|
||||
|
||||
# Build project hierarchy - create all projects in the path
|
||||
current_project_node = None
|
||||
|
||||
for proj_id, proj_name in project_path:
|
||||
if proj_id not in project_trees:
|
||||
project_trees[proj_id] = TreeNode(proj_id, proj_name, 'project', logger=self)
|
||||
|
||||
if current_project_node is None:
|
||||
# This is the root project in the path
|
||||
current_project_node = project_trees[proj_id]
|
||||
else:
|
||||
# Link parent to child in the project hierarchy
|
||||
if project_trees[proj_id] not in current_project_node.children:
|
||||
current_project_node.add_child(project_trees[proj_id])
|
||||
current_project_node = project_trees[proj_id]
|
||||
|
||||
# If no project path found (no hierarchy support), create a simple node
|
||||
if not project_path and project_id:
|
||||
if project_id not in project_trees:
|
||||
project_trees[project_id] = TreeNode(project_id, f"Project-{project_id}", 'project', logger=self)
|
||||
current_project_node = project_trees[project_id]
|
||||
|
||||
# 建構完整的 work package 層級結構
|
||||
wp_nodes = {}
|
||||
root_wp_nodes = set()
|
||||
|
||||
# 首先找出所有根 work package(通過檢查父子關係)
|
||||
for wp_id in wp_entries.keys():
|
||||
wp_path = self._get_work_package_path(wp_id)
|
||||
if wp_path:
|
||||
root_wp_id = wp_path[0][0]
|
||||
root_wp_nodes.add(root_wp_id)
|
||||
|
||||
# 為每個根 work package 建構完整樹狀結構
|
||||
for root_wp_id in root_wp_nodes:
|
||||
if current_project_node:
|
||||
root_wp_node = self._build_complete_work_package_tree(root_wp_id, wp_nodes, wp_entries)
|
||||
if root_wp_node and root_wp_node not in current_project_node.children:
|
||||
current_project_node.add_child(root_wp_node)
|
||||
|
||||
return project_trees
|
||||
|
||||
def _display_tree(self, trees: Dict[str, TreeNode]):
|
||||
"""Display tree structure in the specified format"""
|
||||
# Find root nodes (nodes without parents in our tree structure)
|
||||
root_nodes = []
|
||||
all_nodes = set()
|
||||
|
||||
# Collect all nodes from the trees (including nested children)
|
||||
def collect_all_nodes(node):
|
||||
all_nodes.add(node)
|
||||
for child in node.children:
|
||||
collect_all_nodes(child)
|
||||
|
||||
for node in trees.values():
|
||||
collect_all_nodes(node)
|
||||
|
||||
# Find nodes that don't have parents in the tree
|
||||
for node in all_nodes:
|
||||
if node.parent is None or node.parent not in all_nodes:
|
||||
root_nodes.append(node)
|
||||
|
||||
# Remove duplicates and sort root nodes by total hours (descending)
|
||||
root_nodes = list(set(root_nodes))
|
||||
root_nodes.sort(key=lambda x: x.get_total_hours(), reverse=True)
|
||||
|
||||
print(f"📊 Work Package Time Logging Summary")
|
||||
total_hours = sum(node.get_total_hours() for node in root_nodes)
|
||||
print(f"Total Hours Logged: {total_hours:.2f} hours")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
# Display each root and its subtree
|
||||
for root in root_nodes:
|
||||
self._print_node(root, 0)
|
||||
print()
|
||||
|
||||
def _print_node(self, node: TreeNode, depth: int):
|
||||
"""Print a node and its children with proper indentation"""
|
||||
indent = " " * depth
|
||||
prefix = "P" if node.type == 'project' else "W"
|
||||
|
||||
if node.type == 'work_package' and node.entries:
|
||||
hours_info = f" ({node.hours:.2f}h)"
|
||||
completion_info = f" [完成度: {node.get_completion()}]"
|
||||
print(f"{indent}* {prefix}{node.id} - {node.name}{hours_info}{completion_info}")
|
||||
|
||||
# Show time entries for work packages
|
||||
entry_indent = " " * (depth + 1)
|
||||
for entry in node.entries:
|
||||
# 優先顯示留言,如果留言為空則顯示 no-comment
|
||||
comment_or_default = entry['comment'] if entry['comment'] else "no-comment"
|
||||
print(f"{entry_indent}📝 {entry['date']} | {entry['user']} | {comment_or_default} | {entry['hours']:.2f}h")
|
||||
else:
|
||||
# For projects or work packages without direct entries
|
||||
total_hours = node.get_total_hours()
|
||||
hours_info = f" ({total_hours:.2f}h)" if total_hours > 0 else ""
|
||||
|
||||
# 只有work package才顯示完成度
|
||||
if node.type == 'work_package':
|
||||
completion_info = f" [完成度: {node.get_completion()}]"
|
||||
else:
|
||||
completion_info = ""
|
||||
|
||||
print(f"{indent}* {prefix}{node.id} - {node.name}{hours_info}{completion_info}")
|
||||
|
||||
# Print children
|
||||
for child in sorted(node.children, key=lambda x: x.get_total_hours(), reverse=True):
|
||||
self._print_node(child, depth + 1)
|
||||
|
||||
def _generate_json_output(self, trees: Dict[str, TreeNode], start_date: str, end_date: str) -> Dict:
|
||||
"""Generate JSON format output from tree structure"""
|
||||
# Find root nodes using the same logic as _display_tree
|
||||
root_nodes = []
|
||||
all_nodes = set()
|
||||
|
||||
# Collect all nodes from the trees (including nested children)
|
||||
def collect_all_nodes(node):
|
||||
all_nodes.add(node)
|
||||
for child in node.children:
|
||||
collect_all_nodes(child)
|
||||
|
||||
for node in trees.values():
|
||||
collect_all_nodes(node)
|
||||
|
||||
# Find nodes that don't have parents in the tree
|
||||
for node in all_nodes:
|
||||
if node.parent is None or node.parent not in all_nodes:
|
||||
root_nodes.append(node)
|
||||
|
||||
# Remove duplicates and sort by total hours (descending)
|
||||
root_nodes = list(set(root_nodes))
|
||||
root_nodes.sort(key=lambda x: x.get_total_hours(), reverse=True)
|
||||
|
||||
# Calculate total hours across all projects
|
||||
total_hours = sum(node.get_total_hours() for node in root_nodes)
|
||||
|
||||
return {
|
||||
"summary": {
|
||||
"date_range": f"{start_date} to {end_date}",
|
||||
"total_hours": round(total_hours, 2),
|
||||
"total_projects": len(root_nodes),
|
||||
"generation_time": datetime.now().isoformat()
|
||||
},
|
||||
"projects": [node.to_dict() for node in root_nodes]
|
||||
}
|
||||
|
||||
def fetch_and_display_time_entries(self, start_date: str, end_date: str, output_format: str = 'tree'):
|
||||
"""
|
||||
Fetch time entries for the specified date range and display them.
|
||||
|
||||
Args:
|
||||
start_date: Start date in YYYY-MM-DD format
|
||||
end_date: End date in YYYY-MM-DD format
|
||||
output_format: 'tree' for tree display, 'json' for JSON output, 'both' for both
|
||||
"""
|
||||
# Validate inputs
|
||||
start_dt, end_dt = self._validate_date_range(start_date, end_date)
|
||||
|
||||
print(f"🔍 Fetching work package time entries from {start_date} to {end_date}...")
|
||||
print("=" * 60)
|
||||
|
||||
# Build API request parameters
|
||||
params = self._build_filter_params(start_date, end_date)
|
||||
|
||||
# Fetch time entries
|
||||
try:
|
||||
time_entries = self._fetch_paginated_data(self.time_entries_endpoint, params)
|
||||
print(f"✓ Successfully fetched {len(time_entries)} time entries")
|
||||
print()
|
||||
except APIConnectionError as e:
|
||||
print(f"✗ Failed to fetch time entries: {e}")
|
||||
raise
|
||||
|
||||
if not time_entries:
|
||||
print("⚠ No time entries found for the specified date range")
|
||||
return None if output_format == 'json' else None
|
||||
|
||||
# Build tree structure
|
||||
trees = self._build_tree_structure(time_entries)
|
||||
|
||||
# Handle different output formats
|
||||
if output_format == 'tree':
|
||||
self._display_tree(trees)
|
||||
return None
|
||||
elif output_format == 'json':
|
||||
json_output = self._generate_json_output(trees, start_date, end_date)
|
||||
print(json.dumps(json_output, indent=2, ensure_ascii=False))
|
||||
return json_output
|
||||
elif output_format == 'both':
|
||||
self._display_tree(trees)
|
||||
print("\n" + "="*60)
|
||||
print("JSON Output:")
|
||||
print("="*60)
|
||||
json_output = self._generate_json_output(trees, start_date, end_date)
|
||||
print(json.dumps(json_output, indent=2, ensure_ascii=False))
|
||||
return json_output
|
||||
|
||||
|
||||
def main():
|
||||
"""Command line interface for the work package logger"""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description='Fetch and display work package time entries in tree format')
|
||||
parser.add_argument('start_date', help='Start date (YYYY-MM-DD)')
|
||||
parser.add_argument('end_date', help='End date (YYYY-MM-DD)')
|
||||
parser.add_argument('--test-connection', action='store_true', help='Test API connection only')
|
||||
parser.add_argument('--format', choices=['tree', 'json', 'both'], default='tree',
|
||||
help='Output format: tree (default), json, or both')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
if args.test_connection:
|
||||
logger.test_connection()
|
||||
return
|
||||
|
||||
logger.fetch_and_display_time_entries(args.start_date, args.end_date, args.format)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
exit(main())
|
||||
63
README.md
63
README.md
|
|
@ -1,6 +1,6 @@
|
|||
# Report Generation Skills Experiment
|
||||
|
||||
This repository contains AI skills for automated report generation. Currently includes weekly project report generators that transform time tracking data into formatted management reports.
|
||||
This repository contains AI skills for automated report generation. Currently includes a weekly project report generator that transforms time tracking data into formatted management reports.
|
||||
|
||||
## Available Skills
|
||||
|
||||
|
|
@ -8,39 +8,6 @@ This repository contains AI skills for automated report generation. Currently in
|
|||
|
||||
Automatically generates weekly project reports (項目週報) from exported Excel time tracking data.
|
||||
|
||||
### 📦 Week Work Package Log (`week_work_package_log`)
|
||||
|
||||
Fetches and displays time entries grouped by work packages for specified date ranges via OpenProject API. Provides detailed analysis of work package time logging patterns with user, project, and activity information.
|
||||
|
||||
**Key Features:**
|
||||
- 🌳 **Hierarchical Display**: Shows project and work package relationships in tree format
|
||||
- 📊 **Completion Tracking**: Displays work package completion percentages with intelligent calculation
|
||||
- 📝 **Time Analysis**: Detailed breakdown of time entries by user, activity, and comments
|
||||
- 📊 **Multiple Output Formats**: Tree view for human reading, JSON for data integration
|
||||
|
||||
**Completion Calculation Logic:**
|
||||
1. **Closed/Rejected work packages**: Automatically marked as 100%
|
||||
2. **Explicit completion**: Uses percentageDone or derivedPercentageDone from OpenProject
|
||||
3. **Recursive calculation**: Parent work packages derive completion from children averages
|
||||
4. **No-info handling**: Leaf nodes default to 0%, parents calculate from children
|
||||
|
||||
### 🚀 Week Report Generator from JSON (`week_report_gen_from_json`)
|
||||
|
||||
**New Enhanced Version**: Automatically generates weekly project reports directly from OpenProject API data via JSON format. Features intelligent cell merging, hierarchical project structure recognition, automatic progress tracking, and streamlined workflow.
|
||||
|
||||
**Progress Auto-Fill Logic:**
|
||||
- 🔍 **When C column (子項目名稱) contains work_package**: Automatically fills D column (進度) with work package completion percentage
|
||||
- 📋 **When C column is empty or "-"**: Automatically fills D column with E column work item completion percentage
|
||||
- ⚙️ **Smart Integration**: Seamlessly integrates with `week_work_package_log` completion tracking system
|
||||
|
||||
**Key Advantages:**
|
||||
- 🔄 **Direct API Integration**: No manual Excel export required
|
||||
- 🎯 **Smart Cell Merging**: Automatic B/C/D column merging for cleaner presentation
|
||||
- 📊 **Hierarchical Structure**: Proper project → sub-project → task organization
|
||||
- 📈 **Automatic Progress Tracking**: Auto-fills D column (進度) with work package completion percentages
|
||||
- ⚡ **Automated Workflow**: From data fetch to formatted report in one command
|
||||
- 🎨 **Professional Layout**: Enhanced visual grouping and formatting
|
||||
|
||||
## Getting Started
|
||||
|
||||
### Prerequisites
|
||||
|
|
@ -49,37 +16,9 @@ Fetches and displays time entries grouped by work packages for specified date ra
|
|||
# Create virtual environment and install dependencies
|
||||
uv sync
|
||||
|
||||
# Verify OpenProject API connection
|
||||
uv run --env-file .env python -c 'import os; print(os.getenv("OPENPROJECT_TOKEN"))'
|
||||
```
|
||||
|
||||
### Usage Examples
|
||||
|
||||
#### Original Excel-based Generator
|
||||
```bash
|
||||
uv run --env-file .env python .claude\skills\week_report_gen\generate_report.py "temp\cost-report-2026-01-16-T-16-22-3620260116-7-1r1n4h.xls" "temp\項目週報-智能控制組-20260119.xlsx"
|
||||
```
|
||||
|
||||
#### Work Package Analysis
|
||||
```bash
|
||||
# Tree view with completion percentages
|
||||
uv run --env-file .env python .claude\skills\week_work_package_log\work_package_logger.py "2026-01-22" "2026-01-22" --format tree
|
||||
|
||||
# JSON output for data integration
|
||||
uv run --env-file .env python .claude\skills\week_work_package_log\work_package_logger.py "2026-01-22" "2026-01-22" --format json
|
||||
|
||||
# Both formats
|
||||
uv run --env-file .env python .claude\skills\week_work_package_log\work_package_logger.py "2026-01-22" "2026-01-22" --format both
|
||||
```
|
||||
|
||||
#### New JSON-based Generator (Recommended)
|
||||
```bash
|
||||
# Generate report for specific date range
|
||||
uv run --env-file .env python .claude\skills\week_report_gen_from_json\generate_report.py "2026-01-20" "2026-01-22" "temp\週報-智能控制組-20260123.xlsx"
|
||||
|
||||
# Generate report for current week (default)
|
||||
uv run --env-file .env python .claude\skills\week_report_gen_from_json\generate_report.py
|
||||
|
||||
# Use existing JSON data file
|
||||
uv run --env-file .env python .claude\skills\week_report_gen_from_json\generate_report.py "data.json" "output.xlsx"
|
||||
```
|
||||
|
|
|
|||
192
example.py
192
example.py
|
|
@ -1,151 +1,21 @@
|
|||
"""
|
||||
Example: Various Report Generation Methods
|
||||
Example: Generate Weekly Report
|
||||
|
||||
This script demonstrates how to use different skills:
|
||||
1. week_report_gen: Generate from Excel cost report
|
||||
2. week_work_package_log: Analyze work package time entries with completion tracking
|
||||
3. week_report_gen_from_json: Generate report directly from API data
|
||||
This script demonstrates how to use the week_report_gen skill
|
||||
to generate a weekly report from a cost report Excel file.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Add the skill directory to the path
|
||||
skill_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'skills', 'week_report_gen')
|
||||
sys.path.insert(0, skill_path)
|
||||
|
||||
from generate_report import generate_weekly_report
|
||||
|
||||
|
||||
def main():
|
||||
"""
|
||||
Main function demonstrating various report generation methods.
|
||||
"""
|
||||
print("Report Generation Skills - Examples")
|
||||
print("====================================")
|
||||
print()
|
||||
print("Choose an example to run:")
|
||||
print("1. Work Package Analysis (with completion tracking)")
|
||||
print("2. Weekly Report Generation (from Excel)")
|
||||
print("3. Weekly Report Generation (from JSON/API)")
|
||||
print()
|
||||
|
||||
try:
|
||||
choice = input("Enter your choice (1-3): ").strip()
|
||||
|
||||
if choice == "1":
|
||||
example_work_package_analysis()
|
||||
elif choice == "2":
|
||||
example_weekly_report_excel()
|
||||
elif choice == "3":
|
||||
example_weekly_report_json()
|
||||
else:
|
||||
print("Invalid choice. Please run the script again.")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nExample interrupted by user.")
|
||||
except Exception as e:
|
||||
print(f"\nUnexpected error: {e}")
|
||||
|
||||
|
||||
def example_work_package_analysis():
|
||||
"""
|
||||
Example: Analyze work package time entries with completion tracking
|
||||
"""
|
||||
print("=" * 70)
|
||||
print("Work Package Analysis - Example")
|
||||
print("=" * 70)
|
||||
print()
|
||||
|
||||
# Add the skill directory to the path
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
skill_path = os.path.join(base_dir, '.claude', 'skills', 'week_work_package_log')
|
||||
sys.path.insert(0, skill_path)
|
||||
|
||||
try:
|
||||
from work_package_logger import WeekWorkPackageLogger
|
||||
|
||||
# Initialize logger
|
||||
logger = WeekWorkPackageLogger()
|
||||
|
||||
# Test connection first
|
||||
print("Testing OpenProject API connection...")
|
||||
if not logger.test_connection():
|
||||
print("Failed to connect to OpenProject API. Check your credentials.")
|
||||
return
|
||||
|
||||
# Example date range (replace with actual dates)
|
||||
start_date = "2026-01-22"
|
||||
end_date = "2026-01-22"
|
||||
|
||||
print(f"\nFetching work package data for {start_date} to {end_date}...")
|
||||
print()
|
||||
|
||||
# Display tree format with completion percentages
|
||||
logger.fetch_and_display_time_entries(start_date, end_date, 'tree')
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
print("JSON Output for Integration:")
|
||||
print("=" * 70)
|
||||
|
||||
# Get JSON format for data integration
|
||||
json_data = logger.fetch_and_display_time_entries(start_date, end_date, 'json')
|
||||
|
||||
if json_data:
|
||||
print(f"\nTotal projects analyzed: {json_data['summary']['total_projects']}")
|
||||
print(f"Total hours logged: {json_data['summary']['total_hours']}h")
|
||||
|
||||
# Show completion summary for work packages
|
||||
print("\nWork Package Completion Summary:")
|
||||
for project in json_data['projects']:
|
||||
_print_wp_completion_summary(project, 0)
|
||||
|
||||
print("\n✓ Work package analysis completed successfully!")
|
||||
|
||||
except ImportError as e:
|
||||
print(f"Error importing work_package_logger: {e}")
|
||||
print("Make sure the week_work_package_log skill is properly installed.")
|
||||
except Exception as e:
|
||||
print(f"Error during work package analysis: {e}")
|
||||
|
||||
|
||||
def _print_wp_completion_summary(node, depth=0):
|
||||
"""Helper function to print work package completion summary"""
|
||||
indent = " " * depth
|
||||
|
||||
if node['type'] == 'work_package' and 'completion' in node:
|
||||
print(f"{indent}- {node['name']}: {node['completion']} ({node['total_hours']:.1f}h)")
|
||||
elif node['type'] == 'project':
|
||||
print(f"{indent}📁 {node['name']} ({node['total_hours']:.1f}h)")
|
||||
|
||||
for child in node.get('children', []):
|
||||
_print_wp_completion_summary(child, depth + 1)
|
||||
|
||||
"""
|
||||
Main function demonstrating various report generation methods.
|
||||
"""
|
||||
print("Report Generation Skills - Examples")
|
||||
print("====================================")
|
||||
print()
|
||||
print("Choose an example to run:")
|
||||
print("1. Work Package Analysis (with completion tracking)")
|
||||
print("2. Weekly Report Generation (from Excel)")
|
||||
print("3. Weekly Report Generation (from JSON/API)")
|
||||
print()
|
||||
|
||||
try:
|
||||
choice = input("Enter your choice (1-3): ").strip()
|
||||
|
||||
if choice == "1":
|
||||
example_work_package_analysis()
|
||||
elif choice == "2":
|
||||
example_weekly_report_excel()
|
||||
elif choice == "3":
|
||||
example_weekly_report_json()
|
||||
else:
|
||||
print("Invalid choice. Please run the script again.")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nExample interrupted by user.")
|
||||
except Exception as e:
|
||||
print(f"\nUnexpected error: {e}")
|
||||
|
||||
|
||||
def example_weekly_report_excel():
|
||||
"""
|
||||
Example: Generate a weekly report from the sample cost report.
|
||||
"""
|
||||
|
|
@ -265,49 +135,5 @@ def example_weekly_report_excel():
|
|||
return 1
|
||||
|
||||
|
||||
def example_weekly_report_json():
|
||||
"""
|
||||
Example: Generate weekly report directly from OpenProject API (JSON-based)
|
||||
"""
|
||||
print("=" * 70)
|
||||
print("Weekly Report from JSON - Example")
|
||||
print("=" * 70)
|
||||
print()
|
||||
|
||||
# Add the skill directory to the path
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
skill_path = os.path.join(base_dir, '.claude', 'skills', 'week_report_gen_from_json')
|
||||
sys.path.insert(0, skill_path)
|
||||
|
||||
try:
|
||||
from generate_report import main as generate_json_report
|
||||
|
||||
# Example parameters
|
||||
start_date = "2026-01-20"
|
||||
end_date = "2026-01-22"
|
||||
output_file = os.path.join(base_dir, "temp", "週報-示例-JSON版本.xlsx")
|
||||
|
||||
print(f"Generating report for {start_date} to {end_date}...")
|
||||
print(f"Output file: {output_file}")
|
||||
print()
|
||||
|
||||
# Note: This would typically be called with command line arguments
|
||||
# For demonstration, we're calling it directly
|
||||
print("Note: This example shows the concept. In practice, use:")
|
||||
print(f"uv run --env-file .env python .claude\\skills\\week_report_gen_from_json\\generate_report.py \"{start_date}\" \"{end_date}\" \"{output_file}\"")
|
||||
print()
|
||||
print("Features:")
|
||||
print("- Direct API integration (no manual Excel export)")
|
||||
print("- Intelligent cell merging and formatting")
|
||||
print("- Hierarchical project structure recognition")
|
||||
print("- Work package completion tracking")
|
||||
|
||||
except ImportError as e:
|
||||
print(f"Error importing week_report_gen_from_json: {e}")
|
||||
print("Make sure the week_report_gen_from_json skill is properly installed.")
|
||||
except Exception as e:
|
||||
print(f"Error during JSON report generation: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
sys.exit(main())
|
||||
|
|
|
|||
Loading…
Reference in New Issue