"""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': '分组供后续调度迁移使用；现有脚本仍从首页启动并自行导航，不能直接送到子页面执行。此命令不会操作游戏。', }