#!/usr/bin/env python3 import os, re, json, time from datetime import datetime, timezone from pathlib import Path import requests WORKSPACE = Path('/Users/ahmad/.openclaw/workspace') CRED = WORKSPACE / 'clickup_credentials.md' BASE = 'https://api.clickup.com/api/v2' NEW_SPACE_NAME = 'BigALC OS' FALLBACK_SPACE_NAME = 'Ops' NEW_LISTS = [ 'Today', 'This Week', 'Pipeline', 'Waiting / Blocked', 'Recurring Ops', 'Meetings & Follow-ups', 'Archive', ] def load_token(): env = os.getenv('CLICKUP_API_TOKEN', '').strip() if env: return env txt = CRED.read_text(encoding='utf-8') m = re.search(r'\bpk_[A-Za-z0-9_]+\b', txt) if not m: raise RuntimeError('ClickUp token not found') return m.group(0) class CU: def __init__(self, token): self.h = {'Authorization': token, 'Content-Type': 'application/json'} def req(self, method, path, **kwargs): url = f'{BASE}/{path.lstrip("/")}' r = requests.request(method, url, headers=self.h, timeout=45, **kwargs) if r.status_code >= 400: raise RuntimeError(f'{method} {path} -> {r.status_code}: {r.text[:400]}') if r.text.strip(): return r.json() return {} def teams(self): return self.req('GET', 'team').get('teams', []) def spaces(self, team_id): return self.req('GET', f'team/{team_id}/space').get('spaces', []) def folders(self, space_id): return self.req('GET', f'space/{space_id}/folder').get('folders', []) def folderless_lists(self, space_id): return self.req('GET', f'space/{space_id}/list').get('lists', []) def lists_in_folder(self, folder_id): return self.req('GET', f'folder/{folder_id}/list').get('lists', []) def list_tasks(self, list_id, include_closed=True, max_pages=20): all_t = [] seen_ids = set() for p in range(max_pages): data = self.req('GET', f'list/{list_id}/task', params={'include_closed': str(include_closed).lower(), 'page': str(p)}) tasks = data.get('tasks', []) if not tasks: break new_count = 0 for t in tasks: tid = t.get('id') if tid in seen_ids: continue seen_ids.add(tid) all_t.append(t) new_count += 1 if new_count == 0: break if len(tasks) < 100: break return all_t def create_space(self, team_id, name): payload = { 'name': name, 'multiple_assignees': True, 'features': { 'due_dates': {'enabled': True, 'start_date': True, 'remap_due_dates': True, 'remap_closed_due_date': False}, 'time_tracking': {'enabled': True}, 'tags': {'enabled': True}, 'time_estimates': {'enabled': True}, 'checklists': {'enabled': True}, 'custom_fields': {'enabled': True}, 'remap_dependencies': {'enabled': True}, 'dependency_warning': {'enabled': True}, 'portfolios': {'enabled': True}, } } return self.req('POST', f'team/{team_id}/space', json=payload) def create_list(self, space_id, name): payload = {'name': name, 'content': '', 'due_date': None, 'priority': None, 'assignee': None, 'status': None} return self.req('POST', f'space/{space_id}/list', json=payload) def create_task(self, list_id, payload): return self.req('POST', f'list/{list_id}/task', json=payload) def normalize_title(s): return re.sub(r'\s+', ' ', (s or '').strip().lower()) def target_list_for(task): name = (task.get('name') or '').lower() status = ((task.get('status') or {}).get('status') or '').lower() due = task.get('due_date') tags = [t.get('name','').lower() for t in (task.get('tags') or [])] if 'meeting prep:' in name or 'meeting' in name: return 'Meetings & Follow-ups' if status in ('blocked', 'waiting', 'on hold') or any(x in tags for x in ['blocked', 'waiting']): return 'Waiting / Blocked' if any(x in name for x in ['recurring', 'weekly', 'monthly', 'daily']): return 'Recurring Ops' if status in ('closed', 'done', 'complete'): return 'Archive' if due: try: d = datetime.fromtimestamp(int(due)/1000, tz=timezone.utc) now = datetime.now(timezone.utc) if d.date() == now.date(): return 'Today' if (d - now).days <= 7: return 'This Week' except Exception: pass if status in ('in progress', 'doing', 'active'): return 'This Week' return 'Pipeline' def main(): token = load_token() cu = CU(token) ts = datetime.now().strftime('%Y%m%d_%H%M') backup_dir = WORKSPACE / 'clickup_restructure_backup' / ts backup_dir.mkdir(parents=True, exist_ok=True) teams = cu.teams() if not teams: raise RuntimeError('No ClickUp teams available') team = teams[0] team_id = team['id'] # Inventory snapshot spaces = cu.spaces(team_id) snapshot = {'team': team, 'spaces': []} for sp in spaces: sid = sp['id'] sp_obj = {'space': sp, 'folders': [], 'folderless_lists': [], 'tasks_by_list': {}} folders = cu.folders(sid) for f in folders: fid = f['id'] lists = cu.lists_in_folder(fid) sp_obj['folders'].append({'folder': f, 'lists': lists}) for l in lists: lid = l['id'] sp_obj['tasks_by_list'][lid] = cu.list_tasks(lid, include_closed=True) flists = cu.folderless_lists(sid) sp_obj['folderless_lists'] = flists for l in flists: lid = l['id'] sp_obj['tasks_by_list'][lid] = cu.list_tasks(lid, include_closed=True) snapshot['spaces'].append(sp_obj) (backup_dir / 'inventory.json').write_text(json.dumps(snapshot, indent=2), encoding='utf-8') # Create/find new space (fallback if plan blocks creating new spaces) target_space = next((s for s in spaces if s.get('name') == NEW_SPACE_NAME), None) created_space = False used_fallback_space = False if not target_space: try: target_space = cu.create_space(team_id, NEW_SPACE_NAME) created_space = True except Exception as e: # PAYWALL_004 and similar: cannot create more spaces on current plan fb = next((s for s in spaces if s.get('name') == FALLBACK_SPACE_NAME), None) if not fb: fb = spaces[0] target_space = fb used_fallback_space = True target_space_id = target_space['id'] # Create/find target lists existing_target_lists = {l['name']: l for l in cu.folderless_lists(target_space_id)} created_lists = [] target_lists = {} for ln in NEW_LISTS: if ln in existing_target_lists: target_lists[ln] = existing_target_lists[ln] else: nl = cu.create_list(target_space_id, ln) created_lists.append(nl) target_lists[ln] = nl # Collect source tasks (all lists except those in BigALC OS) source_tasks = [] for sp in snapshot['spaces']: sname = sp['space'].get('name') for lid, tasks in sp['tasks_by_list'].items(): # if already in target space skip list_meta = None for f in sp['folders']: for l in f['lists']: if l['id'] == lid: list_meta = l break if not list_meta: for l in sp['folderless_lists']: if l['id'] == lid: list_meta = l break if sname == NEW_SPACE_NAME: continue for t in tasks: source_tasks.append((sname, list_meta or {}, t)) # Existing task keys in target lists for dedupe existing_keys = set() for ln, lmeta in target_lists.items(): for t in cu.list_tasks(lmeta['id'], include_closed=True): key = (normalize_title(t.get('name')), t.get('due_date') or '') existing_keys.add(key) migrated = [] skipped = [] for sname, lmeta, t in source_tasks: title = t.get('name') or '(untitled)' due = t.get('due_date') or '' key = (normalize_title(title), due) if key in existing_keys: skipped.append({'reason': 'duplicate', 'source_task_id': t.get('id'), 'name': title}) continue tgt_name = target_list_for(t) tgt_list = target_lists[tgt_name] src_link = f"https://app.clickup.com/t/{t.get('id')}" src_desc = t.get('description') or '' status = ((t.get('status') or {}).get('status') or '') desc = ( f"[Migrated from legacy]\\n" f"Source space: {sname}\\n" f"Source list: {lmeta.get('name','unknown')}\\n" f"Source task id: {t.get('id')}\\n" f"Source link: {src_link}\\n" f"Legacy status: {status}\\n\\n" f"Original description:\\n{src_desc}" ) payload = { 'name': title, 'description': desc[:15000], 'due_date': t.get('due_date'), 'start_date': t.get('start_date'), 'priority': (t.get('priority') or {}).get('priority') if t.get('priority') else None, 'tags': list({*(x.get('name') for x in (t.get('tags') or []) if x.get('name')), 'migrated-from-legacy'}), } try: nt = cu.create_task(tgt_list['id'], payload) migrated.append({ 'source_task_id': t.get('id'), 'new_task_id': nt.get('id'), 'name': title, 'target_list': tgt_name, 'new_link': f"https://app.clickup.com/t/{nt.get('id')}" }) existing_keys.add(key) time.sleep(0.08) except Exception as e: skipped.append({'reason': f'create_failed: {e}', 'source_task_id': t.get('id'), 'name': title}) report = [] report.append('# ClickUp Restructure Report') report.append(f'- Timestamp: {datetime.now().isoformat(timespec="seconds")}') report.append(f'- Team: {team.get("name")} ({team_id})') report.append(f'- Backup dir: {backup_dir}') report.append(f'- New space created: {created_space}') report.append(f'- Used fallback existing space: {used_fallback_space}') report.append(f'- Target space: {target_space.get("name")} ({target_space_id})') report.append('') report.append('## Target Lists') for ln in NEW_LISTS: lid = target_lists[ln]['id'] report.append(f'- {ln}: https://app.clickup.com/v/li/{lid}') report.append('') report.append(f'## Migration Summary') report.append(f'- Source tasks scanned: {len(source_tasks)}') report.append(f'- Migrated tasks created: {len(migrated)}') report.append(f'- Skipped: {len(skipped)}') report.append('') report.append('## Migrated (sample first 100)') for m in migrated[:100]: report.append(f"- {m['name']} -> {m['target_list']} | {m['new_link']} (from {m['source_task_id']})") report.append('') report.append('## Skipped / Unresolved (sample first 100)') for s in skipped[:100]: report.append(f"- {s.get('name')} | {s.get('source_task_id')} | {s.get('reason')}") (WORKSPACE / 'clickup_restructure_report.md').write_text('\n'.join(report), encoding='utf-8') (backup_dir / 'migrated.json').write_text(json.dumps(migrated, indent=2), encoding='utf-8') (backup_dir / 'skipped.json').write_text(json.dumps(skipped, indent=2), encoding='utf-8') print(json.dumps({ 'backup_dir': str(backup_dir), 'created_space': created_space, 'used_fallback_space': used_fallback_space, 'target_space_name': target_space.get('name'), 'target_space_id': target_space_id, 'migrated': len(migrated), 'skipped': len(skipped), 'report': str(WORKSPACE / 'clickup_restructure_report.md') }, indent=2)) if __name__ == '__main__': main()