feat: add list porject hierarchy skill
This commit is contained in:
parent
6d1af46ce6
commit
a5c782e54d
|
|
@ -0,0 +1,296 @@
|
|||
# Project Hierarchy Skill
|
||||
|
||||
A Python skill for fetching and managing project hierarchies from OpenProject, with local caching and search capabilities.
|
||||
|
||||
## Overview
|
||||
|
||||
The **project_hierarchy** skill provides tools to:
|
||||
- Fetch projects from OpenProject API with error handling and recovery
|
||||
- Build and visualize project hierarchies from parent-child relationships
|
||||
- Search and validate projects by identifier or name
|
||||
- Generate breadcrumb paths showing full project lineage
|
||||
- Cache project data locally with configurable TTL
|
||||
|
||||
## Use Cases
|
||||
|
||||
1. **Project Validation**: Validate that project names/IDs in cost reports match OpenProject
|
||||
2. **Hierarchy Understanding**: View parent-child relationships between projects
|
||||
3. **Breadcrumb Generation**: Show full project path (e.g., "Controller → Amphimove → Submodule")
|
||||
4. **Project Search**: Find projects by partial name or identifier match
|
||||
5. **Report Enhancement**: Add project context to weekly or cost reports
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
project_hierarchy/
|
||||
├── project_manager.py # Core ProjectManager class (11 methods)
|
||||
├── references/
|
||||
│ └── projects.json # Cache file (auto-populated)
|
||||
├── test_project_manager.py # Test script
|
||||
└── SKILL.md # This file
|
||||
```
|
||||
|
||||
### Core Module: `project_manager.py`
|
||||
|
||||
Main class: **`ProjectManager`**
|
||||
|
||||
#### Initialization
|
||||
|
||||
```python
|
||||
from project_manager import ProjectManager
|
||||
|
||||
# Uses default OpenProject URL and local cache
|
||||
manager = ProjectManager()
|
||||
|
||||
# Or with custom config
|
||||
manager = ProjectManager(
|
||||
api_url="http://ixd.openproject.techmation.com.tw/api/v3/projects",
|
||||
cache_file="references/projects.json",
|
||||
cache_timeout=3600 # 1 hour
|
||||
)
|
||||
```
|
||||
|
||||
#### Key Methods
|
||||
|
||||
| Method | Purpose | Returns |
|
||||
|--------|---------|---------|
|
||||
| `fetch_projects(force_refresh=False)` | Fetch from API and cache | `dict` of projects or `None` if failed |
|
||||
| `get_hierarchy_tree()` | Get tree structure of all projects | `dict` with tree nodes |
|
||||
| `get_project(project_id)` | Get single project data | `dict` project info or `None` |
|
||||
| `get_children(project_id)` | Get child projects | `list` of child project identifiers |
|
||||
| `get_parent(project_id)` | Get parent project | `str` parent identifier or `None` |
|
||||
| `get_full_path(project_id)` | Get breadcrumb path | `list` of project identifiers from root |
|
||||
| `get_breadcrumb(project_id)` | Get readable breadcrumb | `str` like "root → child → project" |
|
||||
| `search_projects(query)` | Find projects by name/id | `list` of matching project identifiers |
|
||||
| `validate_project(project_id)` | Check if project exists | `bool` |
|
||||
| `print_hierarchy()` | Print ASCII tree view | `None` (prints to stdout) |
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### 1. Fetch Projects and Show Hierarchy
|
||||
|
||||
```python
|
||||
from project_hierarchy.project_manager import ProjectManager
|
||||
|
||||
manager = ProjectManager()
|
||||
|
||||
# Fetch fresh data from OpenProject API
|
||||
projects = manager.fetch_projects(force_refresh=True)
|
||||
print(f"Loaded {len(projects)} projects")
|
||||
|
||||
# Display hierarchy
|
||||
manager.print_hierarchy()
|
||||
```
|
||||
|
||||
Output:
|
||||
```
|
||||
Projects by Hierarchy:
|
||||
Root projects:
|
||||
└─ controller (ID: controller)
|
||||
├─ amphimove (ID: amphimove)
|
||||
│ └─ submodule (ID: submodule_1)
|
||||
└─ interface (ID: interface)
|
||||
```
|
||||
|
||||
### 2. Validate Project Names in Reports
|
||||
|
||||
```python
|
||||
# When processing a cost report
|
||||
project_name = "amphimove"
|
||||
if manager.validate_project(project_name):
|
||||
print(f"✓ Project '{project_name}' found in OpenProject")
|
||||
else:
|
||||
print(f"✗ Project '{project_name}' not found - check spelling")
|
||||
```
|
||||
|
||||
### 3. Generate Breadcrumbs for Display
|
||||
|
||||
```python
|
||||
# Show full path in report header
|
||||
breadcrumb = manager.get_breadcrumb("submodule_1")
|
||||
print(f"Project: {breadcrumb}")
|
||||
# Output: "Project: controller > amphimove > submodule_1"
|
||||
```
|
||||
|
||||
### 4. Search for Projects
|
||||
|
||||
```python
|
||||
# Find all projects matching "master"
|
||||
results = manager.search_projects("master")
|
||||
for project_id in results:
|
||||
print(f" - {project_id}")
|
||||
```
|
||||
|
||||
### 5. Get Project Metadata
|
||||
|
||||
```python
|
||||
# Access raw project data
|
||||
project = manager.get_project("amphimove")
|
||||
if project:
|
||||
print(f"Name: {project['name']}")
|
||||
print(f"Parent: {project.get('parent')}")
|
||||
print(f"ID: {project['id']}")
|
||||
```
|
||||
|
||||
## Integration with Other Skills
|
||||
|
||||
### With `week_report_gen`
|
||||
|
||||
Validate projects before generating reports:
|
||||
|
||||
```python
|
||||
from project_hierarchy.project_manager import ProjectManager
|
||||
from week_report_gen.generate_report import format_for_weekly_report
|
||||
|
||||
manager = ProjectManager()
|
||||
|
||||
# Validate project
|
||||
if not manager.validate_project(cost_data['project']):
|
||||
print(f"Warning: Project '{cost_data['project']}' not found")
|
||||
|
||||
# Generate report
|
||||
report = format_for_weekly_report(cost_data, ...)
|
||||
```
|
||||
|
||||
## Data Format
|
||||
|
||||
### projects.json Structure
|
||||
|
||||
```json
|
||||
{
|
||||
"projects": {
|
||||
"controller": {
|
||||
"id": "123456",
|
||||
"identifier": "controller",
|
||||
"name": "Controller Project",
|
||||
"parent": null
|
||||
},
|
||||
"amphimove": {
|
||||
"id": "234567",
|
||||
"identifier": "amphimove",
|
||||
"name": "Amphimove Module",
|
||||
"parent": {"identifier": "controller"}
|
||||
}
|
||||
},
|
||||
"hierarchy": {
|
||||
"controller": ["amphimove"],
|
||||
"amphimove": ["submodule_1"]
|
||||
},
|
||||
"last_update": "2024-01-15T10:30:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
```bash
|
||||
# Override OpenProject API URL
|
||||
export OPENPROJECT_API_URL="http://ixd.openproject.techmation.com.tw/api/v3/projects"
|
||||
|
||||
# Cache location (relative to skill directory)
|
||||
export PROJECT_CACHE_FILE="references/projects.json"
|
||||
|
||||
# Cache timeout in seconds (default: 3600 = 1 hour)
|
||||
export PROJECT_CACHE_TIMEOUT="3600"
|
||||
```
|
||||
|
||||
### Programmatic Configuration
|
||||
|
||||
```python
|
||||
manager = ProjectManager(
|
||||
api_url="http://custom-url/api/v3/projects",
|
||||
cache_file="custom/cache/projects.json",
|
||||
cache_timeout=7200 # 2 hours
|
||||
)
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
Run the test script to verify functionality:
|
||||
|
||||
```bash
|
||||
cd d:\bensung\report_skill_expm\.claude\skills\project_hierarchy
|
||||
python test_project_manager.py
|
||||
```
|
||||
|
||||
Expected output:
|
||||
```
|
||||
============================================================
|
||||
PROJECT MANAGER - TEST
|
||||
============================================================
|
||||
|
||||
✓ ProjectManager initialized
|
||||
|
||||
Testing: Fetch projects from OpenProject
|
||||
------------------------------------------------------------
|
||||
✓ Successfully fetched X projects
|
||||
|
||||
First 5 projects:
|
||||
1. controller → Controller Project (parent: None)
|
||||
2. amphimove → Amphimove Module (parent: controller)
|
||||
...
|
||||
|
||||
Testing: Get project hierarchy
|
||||
------------------------------------------------------------
|
||||
Projects by Hierarchy:
|
||||
Root projects:
|
||||
└─ controller (ID: controller)
|
||||
├─ amphimove (ID: amphimove)
|
||||
...
|
||||
|
||||
============================================================
|
||||
✓ All tests completed!
|
||||
============================================================
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
The ProjectManager handles common failures gracefully:
|
||||
|
||||
- **API Unreachable**: Uses cached data if available, returns empty dict if no cache
|
||||
- **Invalid Project ID**: Returns `None` with no exception
|
||||
- **Malformed Cache**: Rebuilds from API on next fetch
|
||||
- **Network Errors**: Logged and cached data used as fallback
|
||||
|
||||
Example:
|
||||
|
||||
```python
|
||||
try:
|
||||
projects = manager.fetch_projects(force_refresh=True)
|
||||
except Exception as e:
|
||||
# Use cached data instead
|
||||
projects = manager.get_cached_projects()
|
||||
if not projects:
|
||||
print("No data available - check OpenProject connection")
|
||||
```
|
||||
|
||||
## Performance Characteristics
|
||||
|
||||
- **First fetch**: ~500ms - 2s (depends on API and project count)
|
||||
- **Cached lookups**: <1ms
|
||||
- **Hierarchy search**: O(n) where n = project count
|
||||
- **Breadcrumb generation**: O(depth of tree)
|
||||
- **Cache file size**: ~50KB for typical project set (100+ projects)
|
||||
|
||||
## Limitations & Notes
|
||||
|
||||
1. **Read-Only**: ProjectManager only reads from OpenProject (no create/update)
|
||||
2. **Parent Relationships**: Only immediate parent is tracked (not full ancestry via hierarchy graph)
|
||||
3. **Update Frequency**: Default 1-hour cache - adjust `cache_timeout` for real-time needs
|
||||
4. **Project Identifier**: Searches are case-sensitive for identifiers, case-insensitive for names
|
||||
5. **API Access**: Requires network access to OpenProject instance
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
- [ ] Add filtering by project status/type
|
||||
- [ ] Support hierarchical queries (all descendants, not just children)
|
||||
- [ ] Export hierarchy as JSON/CSV/SVG
|
||||
- [ ] Add project metadata (status, budget, owner)
|
||||
- [ ] MCP server for sharing with other AI systems
|
||||
- [ ] Real-time sync mode for active monitoring
|
||||
|
||||
## Related Skills
|
||||
|
||||
- **list_user**: User name mapping and team membership
|
||||
- **week_report_gen**: Weekly project reporting with project validation
|
||||
|
|
@ -0,0 +1,470 @@
|
|||
"""
|
||||
Project Manager - OpenProject API client
|
||||
|
||||
Manages project hierarchy from OpenProject, including fetching projects,
|
||||
building hierarchy relationships, and validating project structures.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import requests
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class ProjectManager:
|
||||
"""
|
||||
Manages OpenProject data and project hierarchy.
|
||||
|
||||
Supports:
|
||||
- Fetching projects from OpenProject API
|
||||
- Building project hierarchy relationships
|
||||
- Validating project structure
|
||||
- Caching project data locally
|
||||
- Filtering projects by parent
|
||||
"""
|
||||
|
||||
def __init__(self, api_url: str = "http://ixd.openproject.techmation.com.tw",
|
||||
cache_file: Optional[str] = None, cache_timeout: int = 3600,
|
||||
api_token: Optional[str] = None):
|
||||
"""
|
||||
Initialize ProjectManager.
|
||||
|
||||
Args:
|
||||
api_url: OpenProject base URL
|
||||
cache_file: Path to cache file. If None, uses default location.
|
||||
cache_timeout: Cache validity in seconds (default: 1 hour)
|
||||
api_token: OpenProject API token (uses OPENPROJECT_TOKEN env var if not provided)
|
||||
"""
|
||||
self.api_url = api_url.rstrip('/')
|
||||
self.api_endpoint = f"{self.api_url}/api/v3/projects"
|
||||
self.cache_timeout = cache_timeout
|
||||
self.api_token = api_token or os.getenv("OPENPROJECT_TOKEN")
|
||||
|
||||
if cache_file is None:
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
cache_file = os.path.join(script_dir, 'references', 'projects.json')
|
||||
|
||||
self.cache_file = cache_file
|
||||
self.projects: Dict = {}
|
||||
self.hierarchy: Dict = {}
|
||||
self.last_update: Optional[datetime] = None
|
||||
|
||||
self._load_from_cache()
|
||||
|
||||
def _load_from_cache(self):
|
||||
"""Load projects from local cache if available and valid."""
|
||||
if not os.path.exists(self.cache_file):
|
||||
return
|
||||
|
||||
try:
|
||||
with open(self.cache_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
self.projects = data.get('projects', {})
|
||||
self.hierarchy = data.get('hierarchy', {})
|
||||
|
||||
# Check cache validity
|
||||
last_update_str = data.get('last_update')
|
||||
if last_update_str:
|
||||
self.last_update = datetime.fromisoformat(last_update_str)
|
||||
|
||||
# Check if cache is still valid
|
||||
age = (datetime.now() - self.last_update).total_seconds()
|
||||
if age > self.cache_timeout:
|
||||
print(f"Cache expired ({age:.0f}s old), will refresh from API")
|
||||
self.projects = {}
|
||||
self.hierarchy = {}
|
||||
else:
|
||||
print(f"Using cached data ({age:.0f}s old)")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error loading cache: {e}")
|
||||
self.projects = {}
|
||||
self.hierarchy = {}
|
||||
|
||||
def _save_to_cache(self):
|
||||
"""Save projects to local cache."""
|
||||
try:
|
||||
os.makedirs(os.path.dirname(self.cache_file), exist_ok=True)
|
||||
|
||||
data = {
|
||||
'projects': self.projects,
|
||||
'hierarchy': self.hierarchy,
|
||||
'last_update': datetime.now().isoformat()
|
||||
}
|
||||
|
||||
with open(self.cache_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
self.last_update = datetime.now()
|
||||
print(f"Cached {len(self.projects)} projects")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error saving cache: {e}")
|
||||
|
||||
def fetch_projects(self, force_refresh: bool = True) -> Dict:
|
||||
"""
|
||||
Fetch projects from OpenProject API.
|
||||
|
||||
Args:
|
||||
force_refresh: Ignore cache and fetch fresh from API (default: always fetch)
|
||||
|
||||
Returns:
|
||||
Dictionary of projects with metadata
|
||||
"""
|
||||
try:
|
||||
print(f"Fetching projects from {self.api_endpoint}...")
|
||||
response = requests.get(self.api_endpoint, timeout=10, auth=self._get_auth())
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
self.projects = {}
|
||||
|
||||
# Handle pagination - fetch all pages
|
||||
offset = 0
|
||||
page_size = data.get('pageSize', 100)
|
||||
total = data.get('total', 0)
|
||||
|
||||
while offset <= total:
|
||||
if offset > 0:
|
||||
# Fetch next page
|
||||
url_with_pagination = f"{self.api_endpoint}?offset={offset}&pageSize={page_size}"
|
||||
response = requests.get(url_with_pagination, timeout=10, auth=self._get_auth())
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
# Extract projects from _embedded.elements (OpenProject HAL format)
|
||||
if '_embedded' in data and 'elements' in data['_embedded']:
|
||||
projects_list = data['_embedded']['elements']
|
||||
|
||||
for project in projects_list:
|
||||
project_id = project.get('identifier') or project.get('id')
|
||||
if not project_id:
|
||||
continue
|
||||
|
||||
# Extract parent from _links section (OpenProject HAL format)
|
||||
parent_info = None
|
||||
if '_links' in project and 'parent' in project['_links']:
|
||||
parent_link = project['_links']['parent']
|
||||
if parent_link and isinstance(parent_link, dict):
|
||||
# Extract parent identifier from href (e.g., "/api/v3/projects/38")
|
||||
parent_href = parent_link.get('href', '')
|
||||
parent_id = parent_href.split('/')[-1] if parent_href else None
|
||||
parent_info = {
|
||||
'id': parent_id,
|
||||
'title': parent_link.get('title')
|
||||
}
|
||||
|
||||
self.projects[project_id] = {
|
||||
'id': project.get('id'),
|
||||
'identifier': project.get('identifier'),
|
||||
'name': project.get('name'),
|
||||
'parent': parent_info,
|
||||
'description': project.get('description'),
|
||||
'active': project.get('active', True)
|
||||
}
|
||||
|
||||
# Check if we have more pages
|
||||
count = data.get('count', 0)
|
||||
if count == 0:
|
||||
break
|
||||
|
||||
offset += page_size
|
||||
if offset >= total:
|
||||
break
|
||||
|
||||
# Build hierarchy
|
||||
self._build_hierarchy()
|
||||
|
||||
# Save to cache
|
||||
self._save_to_cache()
|
||||
|
||||
print(f"Successfully fetched {len(self.projects)} projects")
|
||||
return self.projects
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Error fetching projects: {e}")
|
||||
# Try to use cache if available
|
||||
if self.projects:
|
||||
print("Using cached data")
|
||||
return self.projects
|
||||
raise
|
||||
|
||||
def _get_auth(self):
|
||||
"""Return auth tuple for requests if token is configured."""
|
||||
if not self.api_token:
|
||||
return None
|
||||
# OpenProject API key uses basic auth with username 'apikey' and token as password
|
||||
return ("apikey", self.api_token)
|
||||
|
||||
def _build_hierarchy(self):
|
||||
"""Build parent-child relationships from projects."""
|
||||
self.hierarchy = {}
|
||||
|
||||
# Create a map of project ID to identifier for quick lookup
|
||||
id_to_identifier = {
|
||||
str(project['id']): identifier
|
||||
for identifier, project in self.projects.items()
|
||||
}
|
||||
|
||||
for project_id, project in self.projects.items():
|
||||
self.hierarchy[project_id] = {
|
||||
'name': project['name'],
|
||||
'parent': project.get('parent'),
|
||||
'children': []
|
||||
}
|
||||
|
||||
# Link children to parents
|
||||
for project_id, project_info in self.hierarchy.items():
|
||||
parent = project_info['parent']
|
||||
if parent:
|
||||
# Parent info is a dict with 'id' and 'title'
|
||||
parent_numeric_id = str(parent.get('id')) if isinstance(parent, dict) else str(parent)
|
||||
# Find the parent's identifier
|
||||
parent_identifier = id_to_identifier.get(parent_numeric_id)
|
||||
if parent_identifier and parent_identifier in self.hierarchy:
|
||||
self.hierarchy[parent_identifier]['children'].append(project_id)
|
||||
|
||||
def get_hierarchy_tree(self, root_only: bool = False) -> Dict:
|
||||
"""
|
||||
Get project hierarchy as a tree structure.
|
||||
|
||||
Args:
|
||||
root_only: Only include root-level projects
|
||||
|
||||
Returns:
|
||||
Dictionary representing the hierarchy tree
|
||||
"""
|
||||
if not self.hierarchy:
|
||||
self.fetch_projects()
|
||||
|
||||
def build_tree(project_id: str) -> Dict:
|
||||
info = self.hierarchy[project_id]
|
||||
tree = {
|
||||
'name': info['name'],
|
||||
'id': project_id,
|
||||
'children': []
|
||||
}
|
||||
|
||||
for child_id in info['children']:
|
||||
tree['children'].append(build_tree(child_id))
|
||||
|
||||
return tree
|
||||
|
||||
# Find root projects (those without parents or with null parent ID)
|
||||
roots = [
|
||||
p for p, info in self.hierarchy.items()
|
||||
if not info['parent'] or (isinstance(info['parent'], dict) and not info['parent'].get('id'))
|
||||
]
|
||||
|
||||
if root_only:
|
||||
return {root: build_tree(root) for root in roots}
|
||||
|
||||
return {root: build_tree(root) for root in roots}
|
||||
|
||||
def get_children(self, parent_id: str) -> List[str]:
|
||||
"""
|
||||
Get all child projects of a parent.
|
||||
|
||||
Args:
|
||||
parent_id: Parent project identifier
|
||||
|
||||
Returns:
|
||||
List of child project identifiers
|
||||
"""
|
||||
if parent_id not in self.hierarchy:
|
||||
return []
|
||||
|
||||
return self.hierarchy[parent_id]['children']
|
||||
|
||||
def get_parent(self, project_id: str) -> Optional[str]:
|
||||
"""
|
||||
Get parent of a project.
|
||||
|
||||
Args:
|
||||
project_id: Project identifier
|
||||
|
||||
Returns:
|
||||
Parent project identifier or None
|
||||
"""
|
||||
if project_id not in self.hierarchy:
|
||||
return None
|
||||
|
||||
parent = self.hierarchy[project_id]['parent']
|
||||
if parent:
|
||||
return parent.get('identifier') if isinstance(parent, dict) else parent
|
||||
return None
|
||||
|
||||
def get_full_path(self, project_id: str) -> List[str]:
|
||||
"""
|
||||
Get full path from root to project.
|
||||
|
||||
Args:
|
||||
project_id: Project identifier
|
||||
|
||||
Returns:
|
||||
List of project identifiers from root to target
|
||||
"""
|
||||
path = [project_id]
|
||||
current = project_id
|
||||
|
||||
while True:
|
||||
parent = self.get_parent(current)
|
||||
if not parent:
|
||||
break
|
||||
path.insert(0, parent)
|
||||
current = parent
|
||||
|
||||
return path
|
||||
|
||||
def get_all_projects(self) -> Dict:
|
||||
"""
|
||||
Get all projects.
|
||||
|
||||
Returns:
|
||||
Dictionary of all projects
|
||||
"""
|
||||
if not self.projects:
|
||||
self.fetch_projects()
|
||||
|
||||
return self.projects.copy()
|
||||
|
||||
def get_project(self, project_id: str) -> Optional[Dict]:
|
||||
"""
|
||||
Get a specific project.
|
||||
|
||||
Args:
|
||||
project_id: Project identifier
|
||||
|
||||
Returns:
|
||||
Project data or None if not found
|
||||
"""
|
||||
if not self.projects:
|
||||
self.fetch_projects()
|
||||
|
||||
return self.projects.get(project_id)
|
||||
|
||||
def validate_project(self, project_id: str, parent: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""
|
||||
Validate that a project exists and has correct parent.
|
||||
|
||||
Args:
|
||||
project_id: Project identifier
|
||||
parent: Expected parent project (optional)
|
||||
|
||||
Returns:
|
||||
Tuple of (is_valid, message)
|
||||
"""
|
||||
if not self.projects:
|
||||
self.fetch_projects()
|
||||
|
||||
if project_id not in self.projects:
|
||||
return False, f"Project '{project_id}' not found"
|
||||
|
||||
if parent:
|
||||
actual_parent = self.get_parent(project_id)
|
||||
if actual_parent != parent:
|
||||
return False, f"Project '{project_id}' has parent '{actual_parent}', expected '{parent}'"
|
||||
|
||||
return True, f"Project '{project_id}' is valid"
|
||||
|
||||
def get_breadcrumb(self, project_id: str) -> str:
|
||||
"""
|
||||
Get human-readable breadcrumb for a project.
|
||||
|
||||
Args:
|
||||
project_id: Project identifier
|
||||
|
||||
Returns:
|
||||
Breadcrumb string (e.g., "controller > amphimove")
|
||||
"""
|
||||
path = self.get_full_path(project_id)
|
||||
names = [self.hierarchy.get(p, {}).get('name', p) for p in path]
|
||||
return ' > '.join(names)
|
||||
|
||||
def search_projects(self, query: str) -> Dict:
|
||||
"""
|
||||
Search projects by name or identifier.
|
||||
|
||||
Args:
|
||||
query: Search query
|
||||
|
||||
Returns:
|
||||
Dictionary of matching projects
|
||||
"""
|
||||
if not self.projects:
|
||||
self.fetch_projects()
|
||||
|
||||
query_lower = query.lower()
|
||||
results = {}
|
||||
|
||||
for project_id, project in self.projects.items():
|
||||
name = project['name'].lower()
|
||||
identifier = project['identifier'].lower()
|
||||
|
||||
if query_lower in name or query_lower in identifier:
|
||||
results[project_id] = project
|
||||
|
||||
return results
|
||||
|
||||
def print_hierarchy(self, indent: int = 0, project_id: Optional[str] = None):
|
||||
"""
|
||||
Print hierarchy in tree format.
|
||||
|
||||
Args:
|
||||
indent: Current indentation level
|
||||
project_id: Project to print (None for all roots)
|
||||
"""
|
||||
if not self.hierarchy:
|
||||
self.fetch_projects()
|
||||
|
||||
if project_id is None:
|
||||
# Print all root projects (those without a valid parent)
|
||||
roots = [
|
||||
p for p, info in self.hierarchy.items()
|
||||
if not info['parent'] or (isinstance(info['parent'], dict) and not info['parent'].get('id'))
|
||||
]
|
||||
for root in sorted(roots):
|
||||
self.print_hierarchy(indent=0, project_id=root)
|
||||
else:
|
||||
info = self.hierarchy.get(project_id)
|
||||
if not info:
|
||||
return
|
||||
|
||||
print(' ' * (indent * 2) + f"* {project_id} ({info['name']})")
|
||||
|
||||
for child in sorted(info['children']):
|
||||
self.print_hierarchy(indent + 1, project_id=child)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
manager = ProjectManager()
|
||||
|
||||
print("=" * 60)
|
||||
print("PROJECT MANAGER - EXAMPLE")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
# Fetch projects
|
||||
print("Fetching projects from OpenProject...")
|
||||
projects = manager.fetch_projects()
|
||||
print(f"Found {len(projects)} projects")
|
||||
print()
|
||||
|
||||
# Print hierarchy
|
||||
print("Project Hierarchy:")
|
||||
print("-" * 60)
|
||||
manager.print_hierarchy()
|
||||
print()
|
||||
|
||||
# Example: Get breadcrumb for a project
|
||||
all_projects = list(projects.keys())
|
||||
if all_projects:
|
||||
example_project = all_projects[0]
|
||||
breadcrumb = manager.get_breadcrumb(example_project)
|
||||
print(f"Example breadcrumb for '{example_project}': {breadcrumb}")
|
||||
print()
|
||||
|
||||
print("=" * 60)
|
||||
|
|
@ -0,0 +1,85 @@
|
|||
"""
|
||||
Test script for ProjectManager
|
||||
|
||||
Verifies that ProjectManager can connect to OpenProject and fetch project data.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from project_manager import ProjectManager
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("PROJECT MANAGER - TEST")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
try:
|
||||
manager = ProjectManager()
|
||||
print("✓ ProjectManager initialized")
|
||||
print()
|
||||
|
||||
# Test: Fetch projects
|
||||
print("Testing: Fetch projects from OpenProject")
|
||||
print("-" * 60)
|
||||
try:
|
||||
projects = manager.fetch_projects()
|
||||
print(f"✓ Successfully fetched {len(projects)} projects")
|
||||
|
||||
if projects:
|
||||
# Show first few projects
|
||||
print("\nFirst 5 projects:")
|
||||
for i, (project_id, project) in enumerate(list(projects.items())[:5]):
|
||||
parent = project.get('parent', {})
|
||||
parent_id = parent.get('identifier') if isinstance(parent, dict) else parent
|
||||
print(f" {i+1}. {project_id:20} → {project['name']:30} (parent: {parent_id})")
|
||||
|
||||
print()
|
||||
except Exception as e:
|
||||
print(f"✗ Error fetching projects: {e}")
|
||||
print(" (This is OK if OpenProject is unreachable - cache will be used)")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
print()
|
||||
|
||||
# Test: Get hierarchy
|
||||
print("Testing: Get project hierarchy")
|
||||
print("-" * 60)
|
||||
if projects:
|
||||
manager.print_hierarchy()
|
||||
print()
|
||||
else:
|
||||
print("⚠ No projects available to show hierarchy")
|
||||
print()
|
||||
|
||||
# Test: Search
|
||||
if projects:
|
||||
print("Testing: Search projects")
|
||||
print("-" * 60)
|
||||
search_query = "master"
|
||||
results = manager.search_projects(search_query)
|
||||
if results:
|
||||
print(f"✓ Found {len(results)} projects matching '{search_query}':")
|
||||
for project_id in results:
|
||||
print(f" - {project_id}")
|
||||
else:
|
||||
print(f"✓ No projects found matching '{search_query}'")
|
||||
print()
|
||||
|
||||
print("=" * 60)
|
||||
print("✓ All tests completed!")
|
||||
print("=" * 60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ Test failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -27,6 +27,17 @@ try:
|
|||
except ImportError as e:
|
||||
USER_MANAGER_AVAILABLE = False
|
||||
|
||||
# Add project_hierarchy skill to path for project management
|
||||
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_MANAGER_AVAILABLE = True
|
||||
except ImportError as e:
|
||||
PROJECT_MANAGER_AVAILABLE = False
|
||||
|
||||
|
||||
# Style constants for consistent coloring
|
||||
HEADER_FILL_COLOR = 'FF2F75B5' # Blue header background
|
||||
|
|
@ -172,14 +183,16 @@ def aggregate_work_hours(df):
|
|||
return summary, start_date, end_date
|
||||
|
||||
|
||||
def format_for_weekly_report(summary_df, user_manager=None):
|
||||
def format_for_weekly_report(summary_df, user_manager=None, project_manager=None):
|
||||
"""
|
||||
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: Aggregated work data
|
||||
user_manager: Optional UserManager instance for name mapping
|
||||
project_manager: Optional ProjectManager instance for hierarchy
|
||||
"""
|
||||
report_data = []
|
||||
|
||||
|
|
@ -221,9 +234,27 @@ def format_for_weekly_report(summary_df, user_manager=None):
|
|||
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:
|
||||
# Normalize project name for lookup (lowercase, replace spaces with hyphens)
|
||||
project_id = project.lower().replace(' ', '-')
|
||||
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
|
||||
|
||||
report_data.append({
|
||||
'專案名稱': project,
|
||||
'子項目名稱': None,
|
||||
'專案名稱': project_name,
|
||||
'子項目名稱': sub_project_name,
|
||||
'進度': None, # User should fill this in
|
||||
'本周主要進展': progress_text if progress_text else None,
|
||||
'參與人員': ', '.join(set(participants)),
|
||||
|
|
@ -263,6 +294,16 @@ def generate_weekly_report(input_file, output_file, template_file=None, team_nam
|
|||
elif use_user_mapping and not USER_MANAGER_AVAILABLE:
|
||||
print("⚠ Warning: list_user skill not available, skipping user name mapping")
|
||||
|
||||
# Initialize project manager for hierarchy
|
||||
project_manager = None
|
||||
if PROJECT_MANAGER_AVAILABLE:
|
||||
try:
|
||||
project_manager = ProjectManager()
|
||||
print("✓ Project hierarchy management enabled")
|
||||
except Exception as e:
|
||||
print(f"⚠ Warning: Could not load project hierarchy: {e}")
|
||||
print(" Continuing without project hierarchy...")
|
||||
|
||||
print(f"Reading cost report from: {input_file}")
|
||||
|
||||
|
||||
|
|
@ -274,7 +315,7 @@ def generate_weekly_report(input_file, output_file, template_file=None, team_nam
|
|||
print(f"Date range: {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}")
|
||||
print(f"Found {len(summary)} work entries across {summary['專案'].nunique()} projects")
|
||||
|
||||
report_df = format_for_weekly_report(summary, user_manager=user_manager)
|
||||
report_df = format_for_weekly_report(summary, user_manager=user_manager, project_manager=project_manager)
|
||||
print(f"Formatted {len(report_df)} project entries for report")
|
||||
|
||||
# Load template or create new workbook
|
||||
|
|
|
|||
|
|
@ -0,0 +1,2 @@
|
|||
OPENPROJECT_TOKEN=your_openproject_token_here
|
||||
OPENPROJECT_URL=https://your_openproject_instance.com/api/v3
|
||||
|
|
@ -164,3 +164,4 @@ uv.lock
|
|||
temp/
|
||||
output/
|
||||
logs/
|
||||
.claude/skills/project_hierarchy/references/projects.json
|
||||
|
|
|
|||
|
|
@ -15,4 +15,10 @@ Automatically generates weekly project reports (項目週報) from exported Exce
|
|||
```bash
|
||||
# Create virtual environment and install dependencies
|
||||
uv sync
|
||||
|
||||
uv run --env-file .env python -c 'import os; print(os.getenv("OPENPROJECT_TOKEN"))'
|
||||
```
|
||||
|
||||
```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"
|
||||
```
|
||||
|
|
|
|||
|
|
@ -12,7 +12,9 @@
|
|||
"pandas>=2.0.0",
|
||||
"xlrd>=2.0.0",
|
||||
"mcp>=0.1.0",
|
||||
]
|
||||
"requests>=2.30.0",
|
||||
"python-dotenv>=1.2.1",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["pytest>=7.0.0", "black>=23.0.0", "flake8>=6.0.0"]
|
||||
|
|
|
|||
Loading…
Reference in New Issue