147 lines
7.6 KiB
Python
147 lines
7.6 KiB
Python
"""Configuration-only task audit and development plan. No desktop input."""
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
PAGE_ORDER = ('home', 'shop', 'territory', 'guild', 'any_primary')
|
|
|
|
|
|
class TaskPlanner:
|
|
def __init__(self, data, root):
|
|
self.data = data
|
|
self.root = Path(root)
|
|
self.pages = {page['id']: page for page in data['pages']}
|
|
|
|
def ancestry(self, page_id):
|
|
result = []
|
|
while page_id is not None and page_id not in result:
|
|
if page_id not in self.pages:
|
|
break
|
|
result.append(page_id)
|
|
page_id = self.pages[page_id]['parent']
|
|
return list(reversed(result))
|
|
|
|
def audit(self):
|
|
issues = []
|
|
for key in ('pages', 'edges', 'tasks', 'planned_tasks'):
|
|
ids = [item['id'] for item in self.data[key]]
|
|
issues.extend(f'{key}: duplicate {ident}' for ident, n in Counter(ids).items() if n > 1)
|
|
task_ids = [task['id'] for task in self.data['tasks'] + self.data['planned_tasks']]
|
|
issues.extend(f'tasks: duplicate across lists {ident}'
|
|
for ident, n in Counter(task_ids).items() if n > 1)
|
|
for page in self.data['pages']:
|
|
parent = page['parent']
|
|
if parent is not None and parent not in self.pages:
|
|
issues.append(f"{page['id']}: missing parent {parent}")
|
|
seen, current = set(), page['id']
|
|
while current in self.pages:
|
|
if current in seen:
|
|
issues.append(f"{page['id']}: cyclic parent")
|
|
break
|
|
seen.add(current)
|
|
current = self.pages[current]['parent']
|
|
for variant in page.get('recognition', []):
|
|
for spec in variant:
|
|
if not (self.root / 'assets' / spec['template']).is_file():
|
|
issues.append(f"{page['id']}: missing template {spec['template']}")
|
|
for edge in self.data['edges']:
|
|
if edge['from'] not in self.pages or edge['to'] not in self.pages:
|
|
issues.append(f"{edge['id']}: missing endpoint")
|
|
for task in self.data['tasks'] + self.data['planned_tasks']:
|
|
if task.get('group_page') not in PAGE_ORDER:
|
|
issues.append(f"{task['id']}: unknown group page {task.get('group_page')}")
|
|
page_id = task.get('task_page')
|
|
if page_id is None and task.get('group_page') == 'any_primary':
|
|
continue
|
|
if page_id not in self.pages:
|
|
issues.append(f"{task['id']}: missing task page")
|
|
for ident in self.ancestry(page_id):
|
|
if self.pages[ident].get('automation', {}).get('enabled') is False:
|
|
issues.append(f"{task['id']}: disabled ancestor {ident}")
|
|
script = task.get('script')
|
|
if script and not (self.root / script).is_file():
|
|
issues.append(f"{task['id']}: missing script {script}")
|
|
return {
|
|
'valid': not issues, 'issues': issues,
|
|
'counts': {key: len(self.data[key]) for key in ('pages', 'edges', 'tasks', 'planned_tasks')},
|
|
'recognized_pages': sum(bool(page.get('recognition')) for page in self.pages.values()),
|
|
'verified_edges': sum(edge['status'] in ('live_verified', 'workflow_verified')
|
|
for edge in self.data['edges']),
|
|
}
|
|
|
|
def hub(self, task):
|
|
page = task.get('task_page')
|
|
group = task['group_page']
|
|
if group == 'any_primary':
|
|
return 'any_primary'
|
|
if page and page.startswith('offer.'):
|
|
return 'home.offer'
|
|
ancestors = self.ancestry(page)
|
|
# Siblings share their first entrance below the primary page.
|
|
return ancestors[1] if len(ancestors) > 1 and ancestors[0] == group else page
|
|
|
|
def family(self, task):
|
|
if task.get('reusable_pattern'):
|
|
return task['reusable_pattern']
|
|
ident, page = task['id'], task.get('task_page') or ''
|
|
if 'purchases' in ident or ident == 'shop_daily_coin_ticket':
|
|
return 'priority_purchase'
|
|
if ident in ('sweep', 'planet', 'relic_daily_sweep'):
|
|
return 'sweep'
|
|
if ident == 'wall_command_build':
|
|
return 'explicit_command_build'
|
|
if 'task' in page or ident in ('tasks', 'lizhu', 'pet_challenge_rewards'):
|
|
return 'scrolling_reward_list'
|
|
if ident in ('offer_diamond_fund_rewards', 'offer_relic_pass_rewards', 'offer_pet_pass_rewards'):
|
|
return 'reward_icon_list'
|
|
if 'store' in page or 'free' in ident or ident == 'shop_totem_free_rewards':
|
|
return 'free_offer'
|
|
return 'page_specific'
|
|
|
|
def plan(self, include_deferred=False, include_manual=False, group_page=None):
|
|
rows, omitted = [], []
|
|
existing = {task['id'] for task in self.data['tasks']}
|
|
for task in self.data['tasks'] + self.data['planned_tasks']:
|
|
if group_page and task['group_page'] != group_page:
|
|
continue
|
|
deferred = 'deferred' in task.get('status', '')
|
|
manual = task['id'] in ('sweep', 'wall_command_build') or task.get('frequency') in ('on_demand', 'explicit_user_command')
|
|
if deferred and not include_deferred or manual and not include_manual:
|
|
omitted.append({'id': task['id'], 'reason': 'deferred' if deferred else 'explicit_command'})
|
|
continue
|
|
page_id = task.get('task_page')
|
|
is_existing = task['id'] in existing
|
|
blockers = []
|
|
if not is_existing:
|
|
blockers.append('任务执行器尚未实现')
|
|
if not self.pages.get(page_id, {}).get('recognition'):
|
|
blockers.append('任务页面尚无识别模板')
|
|
if task.get('consumes_resources') is None:
|
|
blockers.append('资源消耗需核实')
|
|
if 'rules_incomplete' in task.get('status', ''):
|
|
blockers.append('业务规则仍待确认')
|
|
if deferred:
|
|
blockers.append('用户要求延期完善')
|
|
rows.append({
|
|
'id': task['id'], 'name': task['name'], 'group_page': task['group_page'],
|
|
'hub': self.hub(task), 'task_page': page_id, 'family': self.family(task),
|
|
'implementation': 'existing_script' if is_existing else 'planned',
|
|
'script': task.get('script'), 'frequency': task.get('frequency'),
|
|
'consumes_resources': task.get('consumes_resources'),
|
|
'requires_explicit_command': manual, 'blockers': blockers,
|
|
'script_start': task.get('current_script_start'),
|
|
'script_finish': task.get('current_script_finish'),
|
|
'navigation_mode': 'legacy_script_owns_navigation' if is_existing else 'future_page_local_task',
|
|
})
|
|
rows.sort(key=lambda row: (PAGE_ORDER.index(row['group_page']), row['hub'] or '',
|
|
row['implementation'] != 'existing_script', row['id']))
|
|
visits = []
|
|
for row in rows:
|
|
if not visits or (visits[-1]['group_page'], visits[-1]['hub']) != (row['group_page'], row['hub']):
|
|
visits.append({'group_page': row['group_page'], 'hub': row['hub'], 'tasks': []})
|
|
visits[-1]['tasks'].append(row['id'])
|
|
return {
|
|
'mode': 'read_only_development_plan', 'tasks': rows, 'visit_groups': visits,
|
|
'omitted': omitted,
|
|
'note': '分组供后续调度迁移使用；现有脚本仍从首页启动并自行导航，不能直接送到子页面执行。此命令不会操作游戏。',
|
|
}
|