Initial RAA parser poster copy

This commit is contained in:
Your Name
2026-08-03 16:18:56 +05:00
commit d01890ff2c
148 changed files with 18924 additions and 0 deletions
+107
View File
@@ -0,0 +1,107 @@
from __future__ import annotations
import re
def normalize_hash_tag(value: str, fallback: str = "source") -> str:
tag = (value or "").strip().lstrip("#").lower()
tag = re.sub(r"\s+", "_", tag)
tag = re.sub(r"[^\wа-яё_]+", "_", tag, flags=re.IGNORECASE)
tag = re.sub(r"_+", "_", tag).strip("_")
return tag or fallback
def strip_trailing_hashtag_line(text: str) -> str:
lines = str(text or "").rstrip().splitlines()
while lines and not lines[-1].strip():
lines.pop()
if not lines:
return ""
parts = lines[-1].split()
if parts and all(part.startswith("#") and normalize_hash_tag(part, "") for part in parts):
lines.pop()
return "\n".join(lines).rstrip()
def publication_hashtags(category_tag: str, source_tag: str) -> str:
tags = [
normalize_hash_tag(category_tag, "category"),
normalize_hash_tag(source_tag, "source"),
]
return " ".join(f"#{tag}" for tag in tags if tag)
import html
def build_publication_text(
text: str,
category_tag: str,
source_tag: str,
format_title: bool = False,
parse_mode: str | None = None,
) -> str:
base = strip_trailing_hashtag_line(text)
lines = base.splitlines()
sep_pattern = re.compile(r"^\s*[━—─\-=\*\#_]{2,}\s*$")
cleaned_lines = []
for line in lines:
if sep_pattern.match(line):
cleaned_lines.append("")
else:
cleaned_lines.append(line)
title_idx = None
for idx, line in enumerate(cleaned_lines):
if line.strip():
title_idx = idx
break
if title_idx is None:
return publication_hashtags(category_tag, source_tag)
formatted_lines = []
for idx, line in enumerate(cleaned_lines):
if idx == title_idx and format_title:
if parse_mode == "html":
escaped_title = html.escape(line.strip())
formatted_lines.append(f"<b>{escaped_title}</b>")
elif parse_mode == "markdown":
formatted_lines.append(f"**{line.strip()}**")
else:
formatted_lines.append(line.strip())
# Ensure a newline break (blank line) after the bold title
if idx + 1 < len(cleaned_lines) and cleaned_lines[idx + 1].strip() != "":
formatted_lines.append("")
else:
if parse_mode == "html":
formatted_lines.append(html.escape(line))
else:
formatted_lines.append(line)
raw_body = "\n".join(formatted_lines)
# Collapse multiple consecutive blank lines to at most two newlines (\n\n)
normalized_body = re.sub(r"\n{3,}", "\n\n", raw_body).strip()
hashtags = publication_hashtags(category_tag, source_tag)
if hashtags:
return f"{normalized_body}\n\n{hashtags}" if normalized_body else hashtags
return normalized_body
def parse_categories(value: object) -> list[str]:
if isinstance(value, list):
raw_items = [str(item) for item in value]
else:
text = str(value or "")
raw_items = re.split(r"[\n,|]+", text)
categories: list[str] = []
seen: set[str] = set()
for item in raw_items:
category = item.strip().strip("#")
key = category.lower()
if category and key not in seen:
categories.append(category)
seen.add(key)
return categories