import json import re from datetime import datetime class ADFBuilder: PANEL_TYPES = {"info", "note", "success", "warning", "error", "tip"} STATUS_COLORS = {"neutral", "purple", "blue", "red", "yellow", "green"} @staticmethod def doc(blocks: list[dict]) -> str: return json.dumps( { "version": 1, "type": "doc", "content": blocks, }, ensure_ascii=False, ) @classmethod def paragraph(cls, text: str) -> dict: return { "type": "paragraph", "content": cls._parse_inline(text), } @classmethod def heading(cls, text: str, level: int = 3) -> dict: return { "type": "heading", "attrs": {"level": min(max(level, 1), 6)}, "content": cls._parse_inline(text), } @staticmethod def code_block(code: str, language: str = "python") -> dict: return { "type": "codeBlock", "attrs": {"language": language}, "content": [{"type": "text", "text": code}], } @classmethod def bullet_list(cls, items: list[str]) -> dict: return { "type": "bulletList", "content": [ { "type": "listItem", "content": [cls.paragraph(item)], } for item in items ], } @classmethod def ordered_list(cls, items: list[str]) -> dict: return { "type": "orderedList", "content": [ { "type": "listItem", "content": [cls.paragraph(item)], } for item in items ], } @staticmethod def mention(account_id: str, display_name: str) -> dict: return { "type": "mention", "attrs": { "id": account_id, "text": f"@{display_name}", "accessLevel": "", "userType": "DEFAULT", }, } @staticmethod def link(url: str, text: str | None = None) -> dict: return { "type": "text", "text": text or url, "marks": [{"type": "link", "attrs": {"href": url}}], } @staticmethod def text_with_marks(text: str, marks: list[dict] | None = None) -> dict: node: dict = {"type": "text", "text": text} if marks: node["marks"] = marks return node @staticmethod def table(headers: list[str], rows: list[list[str]]) -> dict: return { "type": "table", "attrs": {"isNumberColumnEnabled": False, "layout": "default"}, "content": [ { "type": "tableRow", "content": [ { "type": "tableHeader", "attrs": {}, "content": [{"type": "paragraph", "content": [{"type": "text", "text": h}]}], } for h in headers ], } ] + [ { "type": "tableRow", "content": [ { "type": "tableCell", "attrs": {}, "content": [{"type": "paragraph", "content": [{"type": "text", "text": cell}]}], } for cell in row ], } for row in rows ], } @classmethod def panel(cls, content: list[dict], panel_type: str = "info") -> dict: return { "type": "panel", "attrs": {"panelType": panel_type if panel_type in cls.PANEL_TYPES else "info"}, "content": content, } @classmethod def info_panel(cls, text: str) -> dict: return cls.panel( [{"type": "paragraph", "content": [{"type": "text", "text": text}]}], "info", ) @staticmethod def expand(title: str, content: list[dict]) -> dict: return { "type": "expand", "attrs": {"title": title}, "content": content, } @staticmethod def task_list(tasks: list[dict]) -> dict: return { "type": "taskList", "attrs": {}, "content": [ { "type": "taskItem", "attrs": { "localId": task.get("id", f"task-{i}"), "state": task.get("state", "TODO"), }, "content": [ {"type": "text", "text": task.get("text", "")}, ], } for i, task in enumerate(tasks) ], } @staticmethod def blockquote(content: list[dict]) -> dict: return {"type": "blockquote", "content": content} @staticmethod def rule() -> dict: return {"type": "rule"} @classmethod def status(cls, text: str, color: str = "neutral") -> dict: return { "type": "status", "attrs": { "text": text, "color": color if color in cls.STATUS_COLORS else "neutral", }, } @staticmethod def date(iso_date: str) -> dict: ts = int(datetime.fromisoformat(iso_date).timestamp() * 1000) return { "type": "date", "attrs": {"timestamp": ts}, } @staticmethod def media(media_id: str, media_type: str = "file", collection: str = "") -> dict: return { "type": "mediaSingle", "attrs": {"layout": "center"}, "content": [ { "type": "media", "attrs": { "type": media_type, "id": media_id, "collection": collection, }, } ], } @staticmethod def emoji(short_name: str, id: str | None = None, text: str | None = None) -> dict: attrs: dict = {"shortName": short_name} if id: attrs["id"] = id if text: attrs["text"] = text return { "type": "emoji", "attrs": attrs, } @staticmethod def macro(name: str, params: dict | None = None) -> dict: macro_params: dict = {} if params: for key, value in params.items(): macro_params[key] = {"value": value} return { "type": "extension", "attrs": { "extensionType": "com.atlassian.confluence.macro.core", "extensionKey": name, "parameters": {"macroParams": macro_params}, }, } @staticmethod def layout_section(columns: list[list[dict]], column_widths: list[int] | None = None) -> dict: if column_widths is None: column_widths = [100 // len(columns)] * len(columns) return { "type": "layoutSection", "attrs": {"layoutType": "two_equal" if len(columns) == 2 else "default"}, "content": [ { "type": "layoutColumn", "attrs": {"width": column_widths[i]}, "content": col, } for i, col in enumerate(columns) ], } @staticmethod def _parse_inline(text: str) -> list[dict]: pattern = re.compile( r'(\*\*(.+?)\*\*|\*(.+?)\*|`(.+?)`|\[(.+?)\]\((.+?)\))' ) nodes = [] last_end = 0 for m in pattern.finditer(text): if m.start() > last_end: plain = text[last_end:m.start()] if plain: nodes.append({"type": "text", "text": plain}) last_end = m.end() if m.group(2): nodes.append({"type": "text", "text": m.group(2), "marks": [{"type": "strong"}]}) elif m.group(3): nodes.append({"type": "text", "text": m.group(3), "marks": [{"type": "em"}]}) elif m.group(4): nodes.append({"type": "text", "text": m.group(4), "marks": [{"type": "code"}]}) elif m.group(5) and m.group(6): nodes.append({ "type": "text", "text": m.group(5), "marks": [{"type": "link", "attrs": {"href": m.group(6)}}], }) if last_end < len(text): nodes.append({"type": "text", "text": text[last_end:]}) return nodes or [{"type": "text", "text": text}] @classmethod def from_markdown(cls, md_text: str) -> str: blocks = [] lines = md_text.split("\n") i = 0 while i < len(lines): stripped = lines[i].strip() if not stripped: i += 1 continue heading_match = re.match(r"^(#{1,6})\s+(.+)", stripped) if heading_match: level = len(heading_match.group(1)) blocks.append(cls.heading(heading_match.group(2), level)) i += 1 continue if stripped.startswith("```"): lang = stripped[3:].strip() or "text" code_lines = [] i += 1 while i < len(lines) and not lines[i].strip().startswith("```"): code_lines.append(lines[i]) i += 1 i += 1 blocks.append(cls.code_block("\n".join(code_lines), lang)) continue if stripped.startswith("- ") or stripped.startswith("* "): items = [] while i < len(lines) and (lines[i].strip().startswith("- ") or lines[i].strip().startswith("* ")): items.append(lines[i].strip()[2:]) i += 1 blocks.append(cls.bullet_list(items)) continue ordered_match = re.match(r"^(\d+)\.\s+(.+)", stripped) if ordered_match: items = [] while i < len(lines): m = re.match(r"^(\d+)\.\s+(.+)", lines[i].strip()) if not m: break items.append(m.group(2)) i += 1 blocks.append(cls.ordered_list(items)) continue if stripped.startswith("> "): quote_lines = [] while i < len(lines) and lines[i].strip().startswith("> "): quote_lines.append(lines[i].strip()[2:]) i += 1 blocks.append(cls.blockquote([cls.paragraph("\n".join(quote_lines))])) continue if stripped.startswith("---") or stripped.startswith("***"): blocks.append(cls.rule()) i += 1 continue table_match = re.match(r"^\|(.+)\|$", stripped) if table_match and i + 1 < len(lines) and re.match(r"^\|[\s\-:|]+\|$", lines[i + 1].strip()): headers = [h.strip() for h in table_match.group(1).split("|") if h.strip() or h.strip() == ""] rows = [] i += 2 while i < len(lines): row_match = re.match(r"^\|(.+)\|$", lines[i].strip()) if not row_match: break cells = [c.strip() for c in row_match.group(1).split("|")] rows.append(cells) i += 1 blocks.append(cls.table(headers, rows)) continue blocks.append(cls.paragraph(stripped)) i += 1 return cls.doc(blocks) def extract_text_from_adf(adf_json: str) -> str: try: doc = json.loads(adf_json) if isinstance(adf_json, str) else adf_json except (json.JSONDecodeError, TypeError): return "" parts = [] def _walk(node): if isinstance(node, dict): if node.get("type") == "text": parts.append(node.get("text", "")) elif node.get("type") == "mention": parts.append(node.get("attrs", {}).get("text", "@unknown")) elif node.get("type") == "hardBreak": parts.append("\n") for child in node.get("content", []): _walk(child) elif isinstance(node, list): for item in node: _walk(item) _walk(doc) return "".join(parts) ADF_REQUIRED_KEYS = {"version", "type", "content"} def validate_adf(adf_json: str) -> tuple[bool, str]: try: doc = json.loads(adf_json) if isinstance(adf_json, str) else adf_json except (json.JSONDecodeError, TypeError) as e: return False, f"Invalid JSON: {e}" if not isinstance(doc, dict): return False, "ADF root must be a dict" missing = ADF_REQUIRED_KEYS - set(doc.keys()) if missing: return False, f"Missing required keys: {missing}" if doc.get("type") != "doc": return False, f"Root type must be 'doc', got '{doc.get('type')}'" if doc.get("version") != 1: return False, f"ADF version must be 1, got {doc.get('version')}" return True, "OK"