/
dictator
/
ragflow_sync
Обзор
Документация
Войти
/
dictator
/
ragflow_sync
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
master
ttjson2md/core.py
832 строки
32 KB
Developer
refactor: rename ttjson_to_markdown → ttjson2md
22 июн 2026, 14:00
22 июн 2026, 14:00
babc303
Код
Авторство
О чём код?
""" Core conversion functions for TaskTracker JSON to Markdown Converter. This module contains all the parsing and conversion logic for converting TaskTracker JSON documents to Markdown format. """ import json from .attributes import ( format_attribute_value, get_enabled_attributes, ) def apply_text_formatting(text: str, marks: list[dict]) -> str: """Apply text formatting from marks.""" formatted_text = text for mark in marks: mark_type = mark.get("type") if mark_type == "bold": formatted_text = f"**{formatted_text}**" elif mark_type == "italic": formatted_text = f"*{formatted_text}*" elif mark_type == "underline": formatted_text = f"<u>{formatted_text}</u>" elif mark_type == "link": href = mark.get("attrs", {}).get("href", "") if href: formatted_text = f"[{formatted_text}]({href})" elif mark_type == "code": formatted_text = f"`{formatted_text}`" return formatted_text def parse_paragraph(paragraph_data: dict | list) -> str: """Parse paragraph content with formatting.""" content = "" if isinstance(paragraph_data, dict): content_items = paragraph_data.get("content", []) else: content_items = paragraph_data for item in content_items: if isinstance(item, dict) and item.get("type") == "text": text = item.get("text", "") formatted_text = apply_text_formatting(text, item.get("marks", [])) content += formatted_text elif isinstance(item, dict) and item.get("type") == "hardBreak": content += "\n\n" elif isinstance(item, dict) and item.get("type") == "image": src = item.get("attrs", {}).get("src", "") alt = item.get("attrs", {}).get("alt", "Image") content += f"\n\n" return content.strip() def parse_list(list_data: dict, list_type: str = "bullet") -> str: """Parse list content.""" if not isinstance(list_data, dict): return "" items = [] content_items = list_data.get("content", []) for item in content_items: if isinstance(item, dict) and item.get("type") == "listItem": item_content = "" if "content" in item: if isinstance(item["content"], list): item_parts = [] for content_element in item["content"]: if isinstance(content_element, dict): if content_element.get("type") == "paragraph": item_parts.append(parse_paragraph(content_element)) elif content_element.get("type") in [ "orderedList", "bulletList", ]: item_parts.append( parse_list( content_element, content_element.get("type", "bullet"), ) ) else: item_parts.append(parse_paragraph(content_element)) else: item_parts.append(str(content_element)) item_content = "\n".join(item_parts) else: item_content = parse_paragraph(item["content"]) else: item_content = str(item) items.append(item_content) if list_type == "ordered": markdown_items = [f"{i + 1}. {item}" for i, item in enumerate(items)] else: markdown_items = [f"- {item}" for item in items] return "\n".join(markdown_items) def parse_table(table_data: dict) -> str: """Parse table content from JSON structure to Markdown table.""" if not table_data: return "" if table_data.get("type") == "table": return _parse_table_structure(table_data) elif table_data.get("type") == "blocksGrid": return _parse_blocks_grid_structure(table_data) else: return "" def _parse_table_structure(table_data: dict) -> str: """Parse traditional table structure.""" if table_data.get("type") != "table": return "" content = table_data.get("content", []) if not content: return "" rows = [] for row in content: if row.get("type") == "tableRow": cells = [] for cell in row.get("content", []): if cell.get("type") in ["tableHeader", "tableCell"]: cell_content = "" if cell.get("content"): paragraph_texts = [] for item in cell["content"]: if item.get("type") == "paragraph": para_content = "" for para_item in item.get("content", []): if para_item.get("type") == "text": para_content += para_item.get("text", "") elif para_item.get("type") == "hardBreak": para_content += "\n" elif para_item.get( "type" ) == "text" and para_item.get("marks"): text = para_item.get("text", "") formatted_text = apply_text_formatting( text, para_item.get("marks", []) ) para_content += formatted_text paragraph_texts.append(para_content) elif item.get("type") == "heading": heading_content = "" for heading_item in item.get("content", []): if heading_item.get("type") == "text": heading_content += heading_item.get("text", "") paragraph_texts.append(heading_content) elif item.get("type") == "text": paragraph_texts.append(item.get("text", "")) elif item.get("type") in ["bulletList", "orderedList"]: list_content = parse_list( item, item.get("type", "bullet") ) paragraph_texts.append(list_content) cleaned_paragraphs = [] for para in paragraph_texts: cleaned_para = " ".join(para.splitlines()).strip() if cleaned_para: cleaned_paragraphs.append(cleaned_para) cell_content = " ".join(cleaned_paragraphs).strip() cells.append(cell_content.strip()) rows.append(cells) rows = [row for row in rows if any(row)] if not rows: return "" if len(rows) >= 1: headers = rows[0] separator = ["---"] * len(headers) table_lines = [] table_lines.append("| " + " | ".join(headers) + " |") table_lines.append("| " + " | ".join(separator) + " |") for row in rows[1:]: table_lines.append("| " + " | ".join(row) + " |") return "\n".join(table_lines) else: return "" def _parse_blocks_grid_structure(blocks_grid_data: dict) -> str: """Parse blocksGrid structure (non-standard table format).""" if blocks_grid_data.get("type") != "blocksGrid": return "" content = blocks_grid_data.get("content", []) if not content: return "" columns = [] for column in content: if column.get("type") == "column": column_content = [] if "content" in column: for item in column["content"]: if item.get("type") == "paragraph": para_text = "" for para_item in item.get("content", []): if para_item.get("type") == "text": para_text += para_item.get("text", "") elif para_item.get("type") == "hardBreak": para_text += "\n" column_content.append(para_text) elif item.get("type") == "heading": heading_text = "" for heading_item in item.get("content", []): if heading_item.get("type") == "text": heading_text += heading_item.get("text", "") column_content.append(f"## {heading_text}") columns.append("\n".join(column_content)) if len(columns) >= 2: return f"| {columns[0]} | {columns[1]} |" elif len(columns) == 1: return columns[0] else: return "" def _parse_expand_element(expand_data: dict) -> str: """Parse expand element content.""" if expand_data.get("type") != "expand": return "" title = "" if "attrs" in expand_data and "title" in expand_data["attrs"]: title = expand_data["attrs"]["title"] content_parts = [] if "content" in expand_data: for item in expand_data["content"]: if item.get("type") == "expandContent" and "content" in item: for content_item in item["content"]: if content_item.get("type") == "paragraph": content_parts.append(parse_paragraph(content_item)) elif content_item.get("type") == "heading": level = content_item.get("attrs", {}).get("level", 1) content = "" for content_part in content_item.get("content", []): if content_part.get("type") == "text": content += content_part.get("text", "") content_parts.append(f"{'#' * level} {content}") elif content_item.get("type") in ["bulletList", "orderedList"]: content_parts.append( parse_list(content_item, content_item.get("type", "bullet")) ) elif content_item.get("type") in ("table", "blocksGrid"): content_parts.append(parse_table(content_item)) elif content_item.get("type") == "codeBlock": code_content = "" if "content" in content_item: for code_part in content_item["content"]: if code_part.get("type") == "text": code_content += code_part.get("text", "") if code_content: content_parts.append(f"```\n{code_content}\n```") elif content_item.get("type") == "fileLink": file_url = content_item.get("attrs", {}).get("href", "") file_name = content_item.get("attrs", {}).get( "fileName", "file" ) content_parts.append(f"[{file_name}]({file_url})") elif content_item.get("type") == "text": content_parts.append(content_item.get("text", "")) else: content = "" for content_part in content_item.get("content", []): if content_part.get("type") == "text": content += content_part.get("text", "") content_parts.append(content) elif item.get("type") == "paragraph": content_parts.append(parse_paragraph(item)) elif item.get("type") == "heading": level = item.get("attrs", {}).get("level", 1) content = "" for content_part in item.get("content", []): if content_part.get("type") == "text": content += content_part.get("text", "") content_parts.append(f"{'#' * level} {content}") elif item.get("type") in ["bulletList", "orderedList"]: content_parts.append(parse_list(item, item.get("type", "bullet"))) elif item.get("type") == "table" or item.get("type") == "blocksGrid": content_parts.append(parse_table(item)) elif item.get("type") == "codeBlock": code_content = "" if "content" in item: for code_part in item["content"]: if code_part.get("type") == "text": code_content += code_part.get("text", "") if code_content: content_parts.append(f"```\n{code_content}\n```") elif item.get("type") == "fileLink": file_url = item.get("attrs", {}).get("href", "") file_name = item.get("attrs", {}).get("fileName", "file") content_parts.append(f"[{file_name}]({file_url})") elif item.get("type") == "text": content_parts.append(item.get("text", "")) if title: return f"## {title}\n\n" + "\n\n".join(content_parts) else: return "\n\n".join(content_parts) def parse_document_content(content_data: dict | list) -> str: """Parse document content and convert to markdown.""" if not content_data or not isinstance(content_data, dict): return "" if isinstance(content_data, dict) and "content" in content_data: content_blocks = content_data["content"] elif isinstance(content_data, list): content_blocks = content_data else: return str(content_data) markdown_parts = [] for block in content_blocks: if block.get("type") == "paragraph": markdown_parts.append(parse_paragraph(block)) elif block.get("type") == "bulletList": markdown_parts.append(parse_list(block, "bullet")) elif block.get("type") == "orderedList": markdown_parts.append(parse_list(block, "ordered")) elif block.get("type") == "heading": level = block.get("attrs", {}).get("level", 1) content = "" for item in block.get("content", []): if item.get("type") == "text": content += item.get("text", "") markdown_parts.append(f"{'#' * level} {content}") elif block.get("type") in ["table", "blocksGrid"]: table_md = parse_table(block) if table_md: markdown_parts.append(table_md) elif block.get("type") == "expand": expand_md = _parse_expand_element(block) if expand_md: markdown_parts.append(expand_md) else: content = "" for item in block.get("content", []): if item.get("type") == "text": content += item.get("text", "") markdown_parts.append(content) return "\n\n".join(markdown_parts) def parse_nested_pages(data: dict) -> str | None: """Extract and format nested pages from details.childList.""" nested_pages = [] details = data.get("details", {}) if details is None: details = {} child_list = details.get("childList", []) for child in child_list: summary = child.get("summary", "") code = child.get("code", "") if summary and code: nested_pages.append(f"- [{summary}]({code})") return "\n".join(nested_pages) if nested_pages else None def is_tasktracker_format(data: dict) -> bool: """Detect if data is in TaskTracker format.""" required_fields = ["taskId", "title", "description"] return all(field in data for field in required_fields) def is_tql_format(data: dict) -> bool: """Detect if data is from TQL query response.""" return "code" in data and "summary" in data and "description" in data def is_wiki_page_format(data: dict) -> bool: """Detect if data is a wiki_page type TaskTracker ticket.""" suit = data.get("suit", {}) if isinstance(suit, dict): return suit.get("code") == "wiki_page" return False def extract_wiki_page_body(data: dict) -> str: """Extract wiki page body from attributes array or top-level body field.""" # First, check for top-level 'body' field (TQL response format) if "body" in data: value = data["body"] if isinstance(value, str) and value.startswith("{"): try: body_data = json.loads(value) return parse_document_content(body_data) except json.JSONDecodeError: return value elif isinstance(value, dict): return parse_document_content(value) # Second, check attributes array (full ticket details format) attributes = data.get("attributes", []) for attr in attributes: if attr.get("code") == "wiki_page_body": value = attr.get("value", "") if isinstance(value, str) and value.startswith("{"): try: body_data = json.loads(value) return parse_document_content(body_data) except json.JSONDecodeError: return value elif isinstance(value, dict): return parse_document_content(value) return str(value) return "" def extract_tasktracker_fields(data: dict) -> dict: """Extract TaskTracker-specific fields.""" task_id = data.get("taskId") or data.get("id", "") title = data.get("title") or data.get("summary", "Untitled Task") description = data.get("description", "") status = data.get("status", "") priority = data.get("priority", "") assignee = data.get("assignee", "") subtasks = data.get("subtasks", []) children = data.get("children", []) related = data.get("relatedTasks", []) return { "task_id": task_id, "title": title, "description": description, "status": status, "priority": priority, "assignee": assignee, "subtasks": subtasks + children + related, } def extract_tql_fields(data: dict) -> dict: """Extract TQL query response fields.""" task_id = data.get("code", "") title = data.get("summary", "Untitled Task") # Check if this is a wiki_page type - extract body from attributes if is_wiki_page_format(data): description_md = extract_wiki_page_body(data) else: # Description can be: # 1. A JSON string in ProseMirror format # 2. A plain string # 3. descriptionPlain (already extracted plain text) description_raw = data.get("description", "") # Try to parse JSON description if isinstance(description_raw, str) and description_raw.startswith("{"): try: description_data = json.loads(description_raw) description_md = parse_document_content(description_data) except json.JSONDecodeError: description_md = description_raw elif isinstance(description_raw, dict): description_md = parse_document_content(description_raw) else: # Use descriptionPlain if available and description is empty/unusable description_md = data.get("descriptionPlain", description_raw) # Extract other fields suit = data.get("suit", {}) space = data.get("space", {}) status_obj = data.get("status", {}) return { "task_id": task_id, "title": title, "description": description_md, "suit": suit.get("code", "") if isinstance(suit, dict) else str(suit), "space": space.get("code", "") if isinstance(space, dict) else str(space), "status": status_obj.get("name", "") if isinstance(status_obj, dict) else str(status_obj), "created_at": data.get("createdAt", ""), "updated_at": data.get("updatedAt", ""), } def extract_all_attributes(data: dict, include_attrs: set[str] | None = None, exclude_attrs: set[str] | None = None) -> list[tuple[str, str]]: """Extract all attributes from a TaskTracker ticket. Parses both the `attributes` array and top-level fields. Returns a list of (display_name, formatted_value) tuples. Args: data: TaskTracker ticket dict include_attrs: If provided, only include these attribute codes exclude_attrs: If provided, exclude these attribute codes Returns: List of (name, value) tuples for non-empty attributes """ enabled_attrs = get_enabled_attributes(include_attrs, exclude_attrs) results = [] # Build a lookup from attribute code to its value from the attributes array attr_values = {} for attr in data.get("attributes", []): if isinstance(attr, dict) and "code" in attr: attr_values[attr["code"]] = attr # Top-level field mappings: attribute_code -> (value_extractor, type_name) top_level_fields = { "suit": (lambda d: d.get("suit", {}), "suit"), "space": (lambda d: d.get("space", {}), "space"), "created_at": (lambda d: d.get("createdAt", ""), "text"), "updated_at": (lambda d: d.get("updatedAt", ""), "text"), "created_by": (lambda d: d.get("createdBy", {}), "user"), "updated_by": (lambda d: d.get("updatedBy", {}), "user"), } for attr_code, attr_config in enabled_attrs.items(): value = None attr_type = "text" # default type # Check if this is a top-level field if attr_code in top_level_fields: extractor, type_name = top_level_fields[attr_code] raw_value = extractor(data) attr_type = type_name # Special handling for suit and space if attr_code == "suit": if isinstance(raw_value, dict): code = raw_value.get("code", "") name = raw_value.get("name", "") if code or name: value = f"{code} ({name})" if (code and name) else (code or name) elif raw_value: value = str(raw_value) elif attr_code == "space": if isinstance(raw_value, dict): code = raw_value.get("code", "") name = raw_value.get("name", "") if code or name: value = f"{code} ({name})" if (code and name) else (code or name) elif raw_value: value = str(raw_value) else: value = format_attribute_value(raw_value, attr_type) # Check if this is in the attributes array elif attr_code in attr_values: attr_obj = attr_values[attr_code] raw_value = attr_obj.get("value") attr_type = attr_obj.get("type", "text") value = format_attribute_value(raw_value, attr_type) # Add to results if non-empty if value: display_name = attr_config.get("name", attr_code) results.append((display_name, value)) return results def parse_task_hierarchy(tasks: list) -> str | None: """Parse TaskTracker task hierarchy.""" if not tasks: return None task_items = [] for task in tasks: if isinstance(task, dict): task_title = task.get("title", task.get("summary", "Unnamed Task")) task_id = task.get("taskId") or task.get("id", "") if task_title and task_id: task_items.append(f"- [{task_title}]({task_id})") elif task_title: task_items.append(f"- {task_title}") elif isinstance(task, str): task_items.append(f"- {task}") return "\n".join(task_items) if task_items else None def convert_tasktracker_to_markdown(data: dict, include_attrs: set[str] | None = None, exclude_attrs: set[str] | None = None) -> str: """Convert TaskTracker JSON to Markdown with task hierarchy.""" task_fields = extract_tasktracker_fields(data) title = task_fields["title"] description = task_fields["description"] metadata = [] if task_fields["task_id"]: metadata.append(f"# [{task_fields['task_id']}] {title}") else: metadata.append(f"# {title}") # Include all attributes attributes = extract_all_attributes(data, include_attrs, exclude_attrs) for name, value in attributes: metadata.append(f"**{name}:** {value}") content_parts = metadata if description: content_parts.append("## Description") content_parts.append(description) nested_tasks_section = parse_task_hierarchy(task_fields["subtasks"]) result = "\n\n".join(content_parts) if nested_tasks_section: result += f"\n\n# Nested Tasks\n\n{nested_tasks_section}" return result def convert_tql_to_markdown(data: dict, include_attrs: set[str] | None = None, exclude_attrs: set[str] | None = None) -> str: """Convert TQL query response to Markdown.""" fields = extract_tql_fields(data) # Build metadata metadata = [] if fields["task_id"]: metadata.append(f"# [{fields['task_id']}] {fields['title']}") else: metadata.append(f"# {fields['title']}") # For non-wiki pages, include all attributes if not is_wiki_page_format(data): attributes = extract_all_attributes(data, include_attrs, exclude_attrs) for name, value in attributes: metadata.append(f"**{name}:** {value}") # Combine metadata with description content_parts = metadata if fields["description"]: # For non-wiki pages, add Description header if not is_wiki_page_format(data): content_parts.append("## Description") content_parts.append(fields["description"]) return "\n\n".join(content_parts) def convert_json_to_markdown(data: dict | list, include_attrs: set[str] | None = None, exclude_attrs: set[str] | None = None) -> str: """Main conversion function with TaskTracker format detection. Handles single dicts, JSON arrays, and auto-detects format type. """ # If input is a list, convert each item and join if isinstance(data, list): return convert_list_to_markdown(data, include_attrs, exclude_attrs) # Check if this is a TQL query response format if is_tql_format(data): return convert_tql_to_markdown(data, include_attrs, exclude_attrs) # Check if this is a TaskTracker format if is_tasktracker_format(data): return convert_tasktracker_to_markdown(data, include_attrs, exclude_attrs) # Fall back to original behavior for generic JSON title = data.get("info", {}).get("summary", "") body_content = data.get("info", {}).get("body", "{}") if isinstance(body_content, str): try: body_data = json.loads(body_content) except json.JSONDecodeError: content_md = body_content else: content_md = parse_document_content(body_data) else: content_md = parse_document_content(body_content) nested_pages_section = parse_nested_pages(data) result = f"# {title}\n\n{content_md}" if nested_pages_section: result += f"\n\n# Nested pages\n\n{nested_pages_section}" return result def validate_json_structure(data: dict | list) -> dict: """Validate JSON structure and return validation results. Returns a dict with: - valid: bool - format: str ('tasktracker', 'wiki', 'tql', 'unknown') - issues: list of str - fields: dict of detected fields Handles both single tickets (dict) and TQL responses (list). """ issues = [] detected_format = "unknown" # Handle list input (TQL response format) if isinstance(data, list): if len(data) == 0: return { "valid": False, "format": "unknown", "issues": ["Empty list"], "fields": {}, } # Validate first item in the list first_item = data[0] if not isinstance(first_item, dict): return { "valid": False, "format": "unknown", "issues": ["List items must be JSON objects"], "fields": {}, } # Recursively validate the first item result = validate_json_structure(first_item) if result["valid"]: result["format"] = "tql" # Mark as TQL format if item is valid return result if not isinstance(data, dict): return { "valid": False, "format": detected_format, "issues": ["Input is not a JSON object"], "fields": {}, } # Check for TQL format (code + summary from TQL search response) is_tql = is_tql_format(data) if is_tql: detected_format = "tql" # TQL format uses 'code' for task ID and 'summary' for title if not data.get("code"): issues.append("Missing 'code' field (TQL task ID)") if not data.get("summary"): issues.append("Missing 'summary' field (TQL title)") # Check for TaskTracker-like format (partial match) elif any(field in data for field in ["taskId", "title", "description"]): is_full_tasktracker = is_tasktracker_format(data) if is_full_tasktracker: detected_format = "tasktracker" fields = extract_tasktracker_fields(data) if not fields["task_id"]: issues.append("Missing 'taskId' or 'id' field") if not fields["title"]: issues.append("Missing 'title' or 'summary' field") else: # Partial TaskTracker - missing required fields detected_format = "tasktracker" fields = extract_tasktracker_fields(data) if "taskId" not in data and "id" not in data: issues.append("Missing 'taskId' or 'id' field") if "title" not in data and "summary" not in data: issues.append("Missing 'title' or 'summary' field") if "description" not in data: issues.append("Missing 'description' field") elif "info" in data: detected_format = "wiki" info = data.get("info", {}) if not info.get("summary"): issues.append("Missing 'info.summary' field") if not info.get("body"): issues.append("Missing 'info.body' field") else: issues.append("Unrecognized JSON format") return { "valid": len(issues) == 0, "format": detected_format, "issues": issues, "fields": extract_tasktracker_fields(data) if detected_format == "tasktracker" else data.get("info", {}), } def convert_list_to_markdown(data_list: list, include_attrs: set[str] | None = None, exclude_attrs: set[str] | None = None) -> str: """Convert a list of TaskTracker JSON objects to Markdown. Args: data_list: List of TaskTracker JSON dicts include_attrs: If provided, only include these attribute codes exclude_attrs: If provided, exclude these attribute codes Returns: Combined Markdown string with separators between items """ results = [] for i, item in enumerate(data_list): if i > 0: results.append("\n\n---\n\n") results.append(convert_json_to_markdown(item, include_attrs, exclude_attrs)) return "".join(results)