109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""Extract structured metadata from a Telegram video caption.
|
||
|
||
Expected format (fields may appear in any order, with extra blank lines, emoji or
|
||
bullet decorations, and varying whitespace)::
|
||
|
||
Index: 001
|
||
Index: 002
|
||
Title: Class 01 Aman sir.mp4
|
||
Topic: Misc• Grammar Practice
|
||
Batch: English ( SSC MAINS )
|
||
Extracted By: @PUTINxINDIA
|
||
|
||
Parsing is line based, so the value of one field can never swallow the next field.
|
||
If a single line holds several known fields (captions flattened by a forwarder),
|
||
the line is split at each known label.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
from models import ParsedCaption
|
||
|
||
REQUIRED_FIELDS = ("title", "topic")
|
||
|
||
_LABEL = r"(?P<label>index|title|topic|batch|extracted[ \t_]*by)"
|
||
# Optional decoration (emoji, bullets, markdown markers) before the label, then the
|
||
# label, optional spaces and an ASCII or full-width colon.
|
||
_LINE_FIELD = re.compile(rf"^[^\w@]*{_LABEL}[ \t]*[:：]", re.IGNORECASE)
|
||
_INLINE_FIELD = re.compile(rf"(?<=\s)[^\w\s@]*{_LABEL}[ \t]*[:：]", re.IGNORECASE)
|
||
_GENERIC_FIELD = re.compile(r"^(?P<key>[^:：]{1,40}?)[ \t]*[:：][ \t]*(?P<value>.*)$")
|
||
_LABEL_KEYS = {
|
||
"index": "index",
|
||
"title": "title",
|
||
"topic": "topic",
|
||
"batch": "batch",
|
||
"extractedby": "extracted_by",
|
||
}
|
||
_DISPLAY = {"title": "Title", "topic": "Topic", "batch": "Batch", "extracted_by": "Extracted By", "index": "Index"}
|
||
|
||
|
||
def _field_key(label: str) -> str:
|
||
return _LABEL_KEYS[re.sub(r"[\s_]+", "", label.lower())]
|
||
|
||
|
||
def _split_fields(line: str) -> list[tuple[str, str]] | None:
|
||
"""Return ``[(field, value), ...]`` if the line starts with a known label."""
|
||
first = _LINE_FIELD.match(line)
|
||
if not first:
|
||
return None
|
||
marks = [(first.end(), first.group("label"))]
|
||
starts = [first.start("label")]
|
||
for inline in _INLINE_FIELD.finditer(line, first.end()):
|
||
starts.append(inline.start())
|
||
marks.append((inline.end(), inline.group("label")))
|
||
result = []
|
||
for i, (value_start, label) in enumerate(marks):
|
||
is_last = i + 1 == len(marks)
|
||
value = line[value_start : len(line) if is_last else starts[i + 1]].strip()
|
||
if not is_last:
|
||
value = value.rstrip(" \t|,;").strip()
|
||
result.append((_field_key(label), value))
|
||
return result
|
||
|
||
|
||
def parse_caption(text: str | None) -> ParsedCaption:
|
||
raw = text or ""
|
||
parsed = ParsedCaption(raw=raw)
|
||
if not raw.strip():
|
||
parsed.warnings.append("caption is empty")
|
||
parsed.missing = list(REQUIRED_FIELDS)
|
||
return parsed
|
||
|
||
for line in raw.replace("\r\n", "\n").replace("\r", "\n").split("\n"):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
fields = _split_fields(line)
|
||
if fields is None:
|
||
generic = _GENERIC_FIELD.match(line)
|
||
if generic and re.search(r"\w", generic.group("key")):
|
||
parsed.extra.setdefault(generic.group("key").strip(), generic.group("value").strip())
|
||
continue
|
||
for name, value in fields:
|
||
_assign(parsed, name, value)
|
||
|
||
parsed.missing = [name for name in REQUIRED_FIELDS if not getattr(parsed, name)]
|
||
for name in parsed.missing:
|
||
parsed.warnings.append(f"{_DISPLAY[name]} field is missing")
|
||
if not parsed.batch:
|
||
parsed.warnings.append("Batch field is missing")
|
||
return parsed
|
||
|
||
|
||
def _assign(parsed: ParsedCaption, name: str, value: str) -> None:
|
||
if name == "index":
|
||
if value:
|
||
parsed.indexes.append(value)
|
||
else:
|
||
parsed.warnings.append("empty Index field ignored")
|
||
return
|
||
if not value:
|
||
parsed.warnings.append(f"{_DISPLAY[name]} field is empty")
|
||
return
|
||
current = getattr(parsed, name)
|
||
if current is None:
|
||
setattr(parsed, name, value)
|
||
elif current != value:
|
||
parsed.warnings.append(f"duplicate {_DISPLAY[name]} field {value!r} ignored; kept {current!r}")
|